Merge branch 'main' into backup/redis-stream-encoder

Resolve conflicts after PR #790 (zset encoder) merged to main:
- redis_string.go: take main's pendingTTL infrastructure (which
  includes PR #790's r14 Finalize warn predicate plus all prior
  fixes); add stream-specific bits — redisKindStream constant,
  streams map field, NewRedisDB init, HandleTTL redisKindStream
  branch, Finalize flushStreams call. Extracted
  inlineWideColumnTTL helper so HandleTTL stays under cyclop=10
  with the 5th wide-column branch.
- redis_string_test.go: take main's version (already has the
  warnFieldsContain helper and FinalizeWarnsOnOverflowOnly test
  landed via PR #790 r14 mirror).
- redis_stream_test.go: remove TestRedisDB_SetTTLArrivesBeforeRows
  duplicate (now in main via PR #790).
- redis_set.go: take main's PR #790 attribution.

No semantic change beyond what was already in PR #790. Stream
encoder bits remain unchanged from this PR's pre-merge state.

Author: bootjp

.github/workflows/jepsen-test-scheduled.yml

@@ -229,7 +229,28 @@ jobs:
             --host 127.0.0.1
       - name: Dump demo cluster log on failure
         if: failure()
-        run: tail -n 500 /tmp/elastickv-demo.log || true
+        # The previous `tail -n 500` truncated a 3-minute workload's
+        # log down to startup-only lines, making it impossible to
+        # correlate a Jepsen anomaly with the server-side state
+        # (start_ts, commit_ts, raft term, write conflicts, lock-
+        # resolver events). Print head + tail inline so the GH UI
+        # still shows the most recent activity at-a-glance, then
+        # upload the full log as an artifact for offline analysis.
+        run: |
+          echo "=== first 200 lines (startup) ==="
+          head -n 200 /tmp/elastickv-demo.log || true
+          echo "=== last 1000 lines (most recent activity) ==="
+          tail -n 1000 /tmp/elastickv-demo.log || true
+          echo "=== full log line count ==="
+          wc -l /tmp/elastickv-demo.log || true
+      - name: Upload demo cluster log on failure
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: elastickv-demo-log
+          path: /tmp/elastickv-demo.log
+          retention-days: 14
+          if-no-files-found: warn
       - name: Stop demo cluster
         if: always()
         run: |

.github/workflows/jepsen-test.yml

@@ -194,6 +194,33 @@ jobs:
             --partition-count 4 --group-count 6 \
             --drain-time 15 \
             --sqs-ports 63501,63502,63503 --host 127.0.0.1
+      - name: Dump demo cluster logs on failure
+        if: failure()
+        # Same rationale as jepsen-test-scheduled.yml: inline summary
+        # for the GH UI + full per-node logs uploaded as artifact so
+        # any Jepsen anomaly can be traced to the server-side ts/raft
+        # state without re-running the workflow.
+        run: |
+          for node in n1 n2 n3; do
+            log=/tmp/elastickv-demo-${node}.log
+            echo "=== ${node}: first 200 lines (startup) ==="
+            head -n 200 "$log" || true
+            echo "=== ${node}: last 500 lines (most recent activity) ==="
+            tail -n 500 "$log" || true
+            echo "=== ${node}: full log line count ==="
+            wc -l "$log" || true
+          done
+      - name: Upload demo cluster logs on failure
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: elastickv-demo-logs
+          path: |
+            /tmp/elastickv-demo-n1.log
+            /tmp/elastickv-demo-n2.log
+            /tmp/elastickv-demo-n3.log
+          retention-days: 14
+          if-no-files-found: warn
       - name: Stop demo cluster
         if: always()
         run: |

adapter/dynamodb_admin.go

@@ -83,6 +83,16 @@ const (
 // future "delete-only" tier reads consistently across the package.
 func (r AdminRole) canWrite() bool { return r == AdminRoleFull }
 
+// canRead reports whether the role authorises non-destructive but
+// sensitive reads — e.g. SQS AdminPeekQueue, which exposes message
+// bodies / attributes. Both AdminRoleReadOnly and AdminRoleFull
+// satisfy this gate; the zero value (unauthenticated / role-less
+// principal) does not. List / Describe paths use the looser
+// session-auth gate because their output is queue metadata already
+// shown on the SPA list page; peek is divergent because the payload
+// is the message bodies themselves.
+func (r AdminRole) canRead() bool { return r == AdminRoleReadOnly || r == AdminRoleFull }
+
 // AdminPrincipal is the authentication context every admin write
 // entrypoint takes. The adapter re-evaluates authorisation against
 // this principal *itself* — it does not trust the caller to have

adapter/redis.go

@@ -355,12 +355,62 @@ func isTransientLeaderRedisError(err error) bool {
 	if err == nil {
 		return false
 	}
-	return errors.Is(err, ErrLeaderNotFound) ||
+	if errors.Is(err, ErrLeaderNotFound) ||
 		errors.Is(err, ErrNotLeader) ||
 		errors.Is(err, kv.ErrLeaderNotFound) ||
 		errors.Is(err, raftengine.ErrNotLeader) ||
 		errors.Is(err, raftengine.ErrLeadershipLost) ||
-		errors.Is(err, raftengine.ErrLeadershipTransferInProgress)
+		errors.Is(err, raftengine.ErrLeadershipTransferInProgress) {
+		return true
+	}
+	// Suffix fallback for gRPC-wrapped sentinels. When the coordinator
+	// forwards a request to a remote leader via the operational gRPC
+	// service and that leader returns ErrLeaderNotFound, the status
+	// interceptor flattens the error to "rpc error: code = Unknown
+	// desc = leader not found"; the typed sentinel chain is stripped
+	// at the wire boundary, so errors.Is misses it. The Jepsen Redis
+	// workload (scheduled run 26035515694) saw workers crash with
+	// `:prefix :rpc` because the un-prefixed `"rpc error: …"` string
+	// reached Carmine. Match the same closed phrase set
+	// kv.hasTransientLeaderPhrase uses, with the same HasSuffix
+	// guard: free-form Contains would misclassify a user-controlled
+	// key like "key: not leader: write conflict" as transient.
+	return hasTransientLeaderSuffix(err.Error())
+}
+
+// redisLeaderErrorPhrases mirrors kv.leaderErrorPhrases (the kv
+// package keeps it unexported). Any new transient-leader phrase the
+// kv layer treats as retryable should also flip a NOTLEADER prefix
+// on the Redis wire so Carmine's with-exceptions catches it; keep
+// in lockstep.
+var redisLeaderErrorPhrases = []string{
+	"not leader",
+	"leader not found",
+	"leadership lost",
+	"leadership transfer in progress",
+}
+
+// hasTransientLeaderSuffix is the suffix-match fallback for
+// isTransientLeaderRedisError. Suffix — not free-form Contains —
+// because cockroachdb/errors %w-prefix and gRPC status.Errorf's
+// "rpc error: code = X desc = <orig>" both leave the original
+// sentinel text at the END of the composed string; a Contains
+// match would tag a user-controlled key like "key: not leader:
+// conflict" as transient.
+//
+// strings.EqualFold on the trailing slice — rather than
+// strings.ToLower + HasSuffix — avoids allocating a copy of the
+// full message. cockroachdb/errors messages can be multi-KB when
+// they carry a serialized stack trace; this matters on the error
+// path under leader-loss storms.
+func hasTransientLeaderSuffix(msg string) bool {
+	for _, phrase := range redisLeaderErrorPhrases {
+		if len(msg) >= len(phrase) &&
+			strings.EqualFold(msg[len(msg)-len(phrase):], phrase) {
+			return true
+		}
+	}
+	return false
 }
 
 func (c *redisMetricsConn) reset(conn redcon.Conn) {

adapter/redis_error_prefix_test.go

@@ -43,6 +43,28 @@ func TestIsTransientLeaderRedisError(t *testing.T) {
 		{"unrelated error", io.EOF, false},
 		{"nil", nil, false},
 		{"wrong-type-looking error", errors.New("WRONGTYPE op"), false},
+		// Suffix-fallback regression cases. These cover the gRPC-
+		// boundary path: when the coordinator forwards to a remote
+		// leader and the remote returns ErrLeaderNotFound, gRPC
+		// flattens it to "rpc error: code = X desc = leader not
+		// found"; the typed sentinel chain is gone. The Jepsen
+		// Redis workload (scheduled run 26035515694) saw workers
+		// crash with `:prefix :rpc` because this case was not
+		// classified as transient and the raw "rpc error: …"
+		// reached Carmine.
+		{"grpc-wrapped leader not found",
+			errors.New("rpc error: code = Unknown desc = leader not found"), true},
+		{"grpc-wrapped not leader",
+			errors.New("rpc error: code = FailedPrecondition desc = raft engine: not leader"), true},
+		{"grpc-wrapped leadership lost",
+			errors.New("rpc error: code = Aborted desc = raft engine: leadership lost"), true},
+		{"grpc-wrapped leadership transfer",
+			errors.New("rpc error: code = Aborted desc = raft engine: leadership transfer in progress"), true},
+		// Suffix discipline: a user-controlled key in the middle
+		// of the message must NOT trigger a false positive. The kv
+		// suffix matcher pins this exact scenario; mirror it here.
+		{"user key embedding 'not leader' in the middle",
+			errors.New("key: not leader: write conflict"), false},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
@@ -74,6 +96,15 @@ func TestWriteRedisError(t *testing.T) {
 			errors.New("WRONGTYPE op"), "WRONGTYPE op"},
 		{"generic io.EOF untouched",
 			io.EOF, io.EOF.Error()},
+		// Suffix-fallback wire reply regression: the gRPC-wrapped
+		// "rpc error: code = Unknown desc = leader not found" string
+		// (the failure mode behind scheduled run 26035515694) must
+		// gain a NOTLEADER prefix on the Redis wire so Carmine maps
+		// it to `:prefix :notleader` and the upstream
+		// jepsen-io/redis with-exceptions catch fires.
+		{"grpc-wrapped leader-not-found gains NOTLEADER prefix",
+			errors.New("rpc error: code = Unknown desc = leader not found"),
+			"NOTLEADER rpc error: code = Unknown desc = leader not found"},
 		// Regression: address-mapping gap errors (raft leader known
 		// but raft→redis address missing in r.leaderRedis) must be
 		// ERR-prefixed at the source so Carmine maps to :prefix :err
@@ -99,3 +130,45 @@ func TestWriteRedisError(t *testing.T) {
 		})
 	}
 }
+
+// TestHasTransientLeaderSuffix_PinsSentinels closes the gap noted
+// at kv/coordinator.go:529 ("A symmetric pin lives in the adapter
+// test package"): the adapter's redisLeaderErrorPhrases set must
+// stay in sync with the actual .Error() text of every transient-
+// leader sentinel the suffix fallback is meant to catch. If a
+// sentinel ever gets renamed (e.g. raftengine.ErrLeadershipLost
+// becomes "raft engine: leadership lost (xyz)") the kv-side pin
+// fails first, but without this adapter-side pin the adapter's
+// phrase list could drift silently and the NOTLEADER classification
+// would regress to the pre-PR-789 worker-crash failure mode.
+//
+// Each case calls hasTransientLeaderSuffix(sentinel.Error()) and
+// asserts true. Wrapping (errors.Wrap / fmt.Errorf %w) is covered
+// by TestIsTransientLeaderRedisError; this test pins the raw
+// .Error() strings only.
+func TestHasTransientLeaderSuffix_PinsSentinels(t *testing.T) {
+	t.Parallel()
+	cases := []struct {
+		name string
+		msg  string
+	}{
+		{"adapter.ErrLeaderNotFound", ErrLeaderNotFound.Error()},
+		{"adapter.ErrNotLeader", ErrNotLeader.Error()},
+		{"kv.ErrLeaderNotFound", kv.ErrLeaderNotFound.Error()},
+		{"raftengine.ErrNotLeader", raftengine.ErrNotLeader.Error()},
+		{"raftengine.ErrLeadershipLost", raftengine.ErrLeadershipLost.Error()},
+		{"raftengine.ErrLeadershipTransferInProgress",
+			raftengine.ErrLeadershipTransferInProgress.Error()},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			if !hasTransientLeaderSuffix(tc.msg) {
+				t.Fatalf("hasTransientLeaderSuffix(%q) = false; "+
+					"redisLeaderErrorPhrases is out of sync with %s — "+
+					"a sentinel rename slipped through. Update the "+
+					"phrase list in adapter/redis.go to match.", tc.msg, tc.name)
+			}
+		})
+	}
+}