fix(sqs): fail-closed for recognised-but-unresolved partition keys

Round 2 Codex P1 on PR #715: when a partition resolver is
installed, an unresolved partitioned-shape key (unknown queue /
out-of-range partition) silently falls through to
engine.GetRoute(routeKey(rawKey)). For !sqs|... keys, routeKey
collapses to !sqs|route|global, so the engine routes the
misconfiguration to the SQS catalog default group instead of
returning a routing error. During partition-map drift / partial
rollout this would silently mis-route HT-FIFO traffic.

Fix

- kv.PartitionResolver gains RecognisesPartitionedKey(key) — a
  shape-only predicate. Implementations answer purely on prefix /
  structural inspection so the router can use it independently
  of the in-memory mapping.
- ShardRouter.ResolveGroup: after ResolveGroup returns ok=false,
  consult RecognisesPartitionedKey. If the resolver recognises the
  shape, return (0, false) WITHOUT engine fallback — the caller
  surfaces a routing error. If the shape is not recognised, fall
  through to the engine as before.
- adapter.SQSPartitionResolver: implements RecognisesPartitionedKey
  via parsePartitionedSQSKey (the same parser ResolveGroup uses).
  Nil-receiver and empty-key return false.

Tests

- kv/shard_router_partition_test.go: new
  TestShardRouter_FailClosedOnRecognisedButUnresolved — pins that
  a recognised-but-unresolved partitioned key returns (0, false)
  AND that a non-recognised key still falls through to the engine.
- adapter/sqs_partition_resolver_test.go: new
  TestSQSPartitionResolver_RecognisesPartitionedKey covers 8
  shape cases (data/vis/byage families with known + unknown
  queues, OOR partition, legacy SQS, queue meta, non-SQS, empty,
  nil). New TestSQSPartitionResolver_RecognisesPartitionedKey_NilReceiver
  pins the typed-nil-safe branch.
- Renamed and expanded
  TestSQSPartitionResolver_UnknownQueueRecognisedButUnresolved /
  TestSQSPartitionResolver_OutOfRangePartitionRecognisedButUnresolved
  to assert RecognisesPartitionedKey == true, pinning the router-
  side fail-closed contract.
- Existing fakePartitionResolver / stubResolver / recordingResolver
  test doubles get the new method (the kv-internal tests don't
  depend on the adapter resolver).

Author: bootjp

adapter/sqs_partition_resolver.go

@@ -68,6 +68,14 @@ var sqsResolverFamilyPrefixes = [][]byte{
 // family prefix (legacy SQS, KV, S3, DynamoDB, queue-meta records,
 // …) so kv.ShardRouter falls through to its byte-range engine for
 // default routing.
+//
+// Returns (0, false) for a partitioned-shaped key whose queue is
+// not in the routes map or whose partition index is beyond
+// len(routes[queue]). The router pairs this with
+// RecognisesPartitionedKey to fail closed instead of falling
+// through — silently routing through the engine's
+// !sqs|route|global default would mis-route HT-FIFO traffic during
+// partition-map drift (codex P1 round 2 on PR #715).
 func (r *SQSPartitionResolver) ResolveGroup(key []byte) (uint64, bool) {
 	if r == nil || len(key) == 0 {
 		return 0, false
@@ -90,6 +98,26 @@ func (r *SQSPartitionResolver) ResolveGroup(key []byte) (uint64, bool) {
 	return groups[partition], true
 }
 
+// RecognisesPartitionedKey reports whether key has the structural
+// shape of a partitioned-SQS key — a partitioned family prefix
+// followed by an encoded queue segment, a '|' terminator, and a
+// fixed-width uint32 partition index. Implementations of
+// kv.PartitionResolver must answer purely on shape so the router
+// can use the predicate to decide between fall-through (not
+// partitioned) and fail-closed (partitioned but unresolved); see
+// kv.PartitionResolver doc and codex P1 round 2 on PR #715.
+//
+// A nil receiver returns false so kv.ShardRouter's typed-nil case
+// (ResolveGroup(nil) == (0, false)) pairs with an honest "I don't
+// recognise anything" answer instead of falsely claiming a shape.
+func (r *SQSPartitionResolver) RecognisesPartitionedKey(key []byte) bool {
+	if r == nil || len(key) == 0 {
+		return false
+	}
+	_, _, ok := parsePartitionedSQSKey(key)
+	return ok
+}
+
 // parsePartitionedSQSKey extracts the (queue, partition) pair from
 // a partitioned-SQS key. Returns ok=false for any key that does not
 // match a partitioned family prefix or that has a malformed queue /

adapter/sqs_partition_resolver_test.go

@@ -147,13 +147,14 @@ func TestSQSPartitionResolver_LegacyKeyFallsThrough(t *testing.T) {
 	}
 }
 
-// TestSQSPartitionResolver_UnknownQueueFallsThrough pins that a
-// well-formed partitioned key for a queue that is NOT in the
-// partition map returns (0, false). This shape can only happen if
-// the cluster lost partition-map entries between writer and reader
-// (an operational misconfiguration); routing through the byte-range
-// engine is the correct fail-closed behaviour.
-func TestSQSPartitionResolver_UnknownQueueFallsThrough(t *testing.T) {
+// TestSQSPartitionResolver_UnknownQueueRecognisedButUnresolved
+// pins the round-5 fail-closed contract: a well-formed partitioned
+// key for a queue that is NOT in the partition map returns
+// ResolveGroup=(0, false) AND RecognisesPartitionedKey=true. The
+// router pairs these so the request fails closed — engine
+// fall-through would silently route the misconfiguration through
+// the SQS catalog default group (codex round-2 P1 on PR #715).
+func TestSQSPartitionResolver_UnknownQueueRecognisedButUnresolved(t *testing.T) {
 	t.Parallel()
 	r := NewSQSPartitionResolver(map[string][]uint64{
 		"known.fifo": {10, 11},
@@ -162,15 +163,17 @@ func TestSQSPartitionResolver_UnknownQueueFallsThrough(t *testing.T) {
 	gid, ok := r.ResolveGroup(key)
 	require.False(t, ok)
 	require.Equal(t, uint64(0), gid)
+	require.True(t, r.RecognisesPartitionedKey(key),
+		"unknown-queue partitioned key must still be recognised — "+
+			"the router pairs Recognised=true with ok=false to fail "+
+			"closed instead of falling through to the engine")
 }
 
-// TestSQSPartitionResolver_OutOfRangePartitionFallsThrough pins
-// that a partition value beyond the configured PartitionCount
-// returns (0, false). Same fail-closed argument as the unknown-
-// queue case — if a writer produced a key with partition 4 but the
-// resolver's view only has partitions [0, 3], routing through the
-// engine surfaces the disagreement at the request boundary.
-func TestSQSPartitionResolver_OutOfRangePartitionFallsThrough(t *testing.T) {
+// TestSQSPartitionResolver_OutOfRangePartitionRecognisedButUnresolved
+// pins the round-5 fail-closed contract for a partition value
+// beyond the configured PartitionCount. Same router-side
+// fail-closed argument as the unknown-queue case.
+func TestSQSPartitionResolver_OutOfRangePartitionRecognisedButUnresolved(t *testing.T) {
 	t.Parallel()
 	r := NewSQSPartitionResolver(map[string][]uint64{
 		"q.fifo": {10, 11},
@@ -182,6 +185,10 @@ func TestSQSPartitionResolver_OutOfRangePartitionFallsThrough(t *testing.T) {
 	gid, ok := r.ResolveGroup(key)
 	require.False(t, ok)
 	require.Equal(t, uint64(0), gid)
+	require.True(t, r.RecognisesPartitionedKey(key),
+		"OOR-partition key must still be recognised — the router "+
+			"pairs Recognised=true with ok=false to fail closed "+
+			"instead of falling through to the engine")
 }
 
 // TestSQSPartitionResolver_NilReceiverIsSafe pins defensive
@@ -197,6 +204,82 @@ func TestSQSPartitionResolver_NilReceiverIsSafe(t *testing.T) {
 	require.Equal(t, uint64(0), gid)
 }
 
+// TestSQSPartitionResolver_RecognisesPartitionedKey pins the
+// shape-only predicate the router uses to decide between
+// fall-through and fail-closed (codex round-2 P1 on PR #715).
+// RecognisesPartitionedKey MUST answer purely on the structural
+// shape — partitioned family prefix + queue + '|' terminator +
+// be32 partition — independent of whether the queue is in the
+// routes map. Otherwise the router could not reliably fail-closed
+// for unresolved-but-recognised keys.
+func TestSQSPartitionResolver_RecognisesPartitionedKey(t *testing.T) {
+	t.Parallel()
+	r := NewSQSPartitionResolver(map[string][]uint64{
+		"known.fifo": {10, 11},
+	})
+	cases := []struct {
+		name string
+		key  []byte
+		want bool
+	}{
+		{
+			name: "data family + known queue",
+			key:  sqsPartitionedMsgDataKey("known.fifo", 0, 1, "id"),
+			want: true,
+		},
+		{
+			name: "vis family + unknown queue (recognised, unresolved)",
+			key:  sqsPartitionedMsgVisKey("not-in-routes.fifo", 0, 1, 1, "id"),
+			want: true,
+		},
+		{
+			name: "byage family + OOR partition (recognised, unresolved)",
+			key:  sqsPartitionedMsgByAgeKey("known.fifo", 99, 1, 1, "id"),
+			want: true,
+		},
+		{
+			name: "legacy SQS key — not partitioned",
+			key:  sqsMsgDataKey("known.fifo", 1, "id"),
+			want: false,
+		},
+		{
+			name: "non-SQS key",
+			key:  []byte("/foo/bar"),
+			want: false,
+		},
+		{
+			name: "queue meta key",
+			key:  sqsQueueMetaKey("known.fifo"),
+			want: false,
+		},
+		{
+			name: "empty",
+			key:  []byte{},
+			want: false,
+		},
+		{
+			name: "nil",
+			key:  nil,
+			want: false,
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			require.Equal(t, tc.want, r.RecognisesPartitionedKey(tc.key))
+		})
+	}
+}
+
+// TestSQSPartitionResolver_RecognisesPartitionedKey_NilReceiver
+// pins the typed-nil-safe branch — same as ResolveGroup, but for
+// the new predicate.
+func TestSQSPartitionResolver_RecognisesPartitionedKey_NilReceiver(t *testing.T) {
+	t.Parallel()
+	var r *SQSPartitionResolver
+	require.False(t, r.RecognisesPartitionedKey([]byte("any-key")))
+}
+
 // TestSQSPartitionResolver_PrefixesAlign pins that the resolver's
 // family-prefix list matches the constants in sqs_keys.go exactly.
 // A future renamed prefix or added family that touches sqs_keys.go

kv/shard_router.go

@@ -21,9 +21,28 @@ import (
 // Implementations must be safe for concurrent use — ResolveGroup is
 // called on the request hot path. Returning (0, false) for a key
 // the resolver does not recognise lets the router fall through to
-// the engine.
+// the engine. Returning (0, false) for a key the resolver DOES
+// recognise (the partitioned shape matches but the queue is not
+// in the map, or the partition is out of range) is also valid;
+// the router uses RecognisesPartitionedKey to distinguish "not
+// partitioned, fall through" from "partitioned but unresolved,
+// fail closed".
+//
+// The fail-closed split matters under partition-map drift / a
+// partial rollout: without it, an unresolved partitioned key
+// would silently land on the engine's SQS-catalog default group
+// (because routeKey normalises every !sqs|... key to
+// !sqs|route|global) instead of surfacing a routing error.
 type PartitionResolver interface {
 	ResolveGroup(key []byte) (uint64, bool)
+
+	// RecognisesPartitionedKey reports whether the key SHAPE is
+	// one this resolver is responsible for. Implementations
+	// answer based on prefix / structural inspection only — the
+	// answer must NOT depend on any in-memory mapping that could
+	// drift out of sync, otherwise the router cannot reliably
+	// fail-closed for unresolved-but-recognised keys.
+	RecognisesPartitionedKey(key []byte) bool
 }
 
 // ShardRouter routes requests to multiple raft groups based on key ranges.
@@ -87,6 +106,15 @@ func (s *ShardRouter) WithPartitionResolver(r PartitionResolver) *ShardRouter {
 // legacy routing (catalog → !sqs|route|global → default group)
 // stays unchanged.
 //
+// Fail-closed for recognised-but-unresolved keys: when the resolver
+// recognises a partitioned shape (RecognisesPartitionedKey == true)
+// but cannot resolve the queue/partition pair (ResolveGroup returns
+// ok=false), the router refuses to fall through to the engine.
+// Otherwise the engine would route the partitioned key to
+// !sqs|route|global's default group, silently mis-routing HT-FIFO
+// traffic during partition-map drift or partial rollout (codex P1
+// round 2 on PR #715).
+//
 // Returns (0, false) when neither the resolver nor the engine
 // recognises the key. Caller surfaces this as an "unknown group"
 // error so a partitioned-prefix key whose queue is missing from the
@@ -97,6 +125,14 @@ func (s *ShardRouter) ResolveGroup(rawKey []byte) (uint64, bool) {
 		if gid, ok := s.partitionResolver.ResolveGroup(rawKey); ok {
 			return gid, true
 		}
+		if s.partitionResolver.RecognisesPartitionedKey(rawKey) {
+			// Partitioned shape, but the resolver cannot map it
+			// (unknown queue, out-of-range partition). Fail closed
+			// — the engine's catalog-level default route would
+			// silently misroute this through routeKey's
+			// !sqs|route|global collapse.
+			return 0, false
+		}
 	}
 	// Engine routes against the user-key view of the byte-range
 	// space; routeKey may rewrite SQS / DynamoDB / Redis-internal

kv/shard_router_partition_test.go

@@ -17,15 +17,31 @@ import (
 // router-side tests — keeps the tests free of any adapter-package
 // dependency so the routing contract is verified at the kv layer in
 // isolation.
+//
+// recognisedPrefix lets a test simulate the "recognised but
+// unresolved" failure mode (codex round-2 P1 on PR #715): keys
+// that start with this prefix and miss the routes map cause
+// ResolveGroup to return (0, false) AND
+// RecognisesPartitionedKey to return true, so the router fails
+// closed instead of falling through to the engine.
 type fakePartitionResolver struct {
-	routes map[string]uint64
+	routes           map[string]uint64
+	recognisedPrefix []byte
 }
 
 func (f *fakePartitionResolver) ResolveGroup(key []byte) (uint64, bool) {
 	gid, ok := f.routes[string(key)]
 	return gid, ok
 }
 
+func (f *fakePartitionResolver) RecognisesPartitionedKey(key []byte) bool {
+	if len(f.recognisedPrefix) == 0 {
+		return false
+	}
+	return len(key) >= len(f.recognisedPrefix) &&
+		bytes.HasPrefix(key, f.recognisedPrefix)
+}
+
 // TestShardRouter_PartitionResolverWins pins that when the resolver
 // claims a key, the router dispatches to the resolver's group even
 // if the byte-range engine would have routed elsewhere. This is the
@@ -176,6 +192,45 @@ func TestShardRouter_ResolverSeesRawKeyNotNormalized(t *testing.T) {
 			"resolver, contrary to the §3.D PR 4-B-2 design")
 }
 
+// TestShardRouter_FailClosedOnRecognisedButUnresolved pins the
+// codex round-2 P1 fix on PR #715: a key the resolver recognises
+// (partitioned shape) but cannot resolve (unknown queue / OOR
+// partition) must NOT fall through to the engine. Without the
+// fail-closed branch, the engine's routeKey-collapsed
+// !sqs|route|global default would silently swallow the misroute
+// during partition-map drift / partial rollout.
+func TestShardRouter_FailClosedOnRecognisedButUnresolved(t *testing.T) {
+	t.Parallel()
+	e := distribution.NewEngine()
+	e.UpdateRoute([]byte(""), nil, 1) // engine would route everything to group 1
+
+	router := NewShardRouter(e)
+	// Resolver claims nothing but RECOGNISES the partitioned shape
+	// for any key starting with "!sqs|msg|data|p|".
+	router.WithPartitionResolver(&fakePartitionResolver{
+		routes:           map[string]uint64{},
+		recognisedPrefix: []byte("!sqs|msg|data|p|"),
+	})
+
+	// A partitioned-shaped key the resolver recognises but cannot
+	// resolve. Without fail-closed this would silently route to
+	// group 1 via the engine's default.
+	_, ok := router.ResolveGroup([]byte("!sqs|msg|data|p|unknown-queue"))
+	require.False(t, ok,
+		"recognised-but-unresolved partitioned key must fail closed; "+
+			"engine fall-through would silently mis-route HT-FIFO "+
+			"traffic to the !sqs|route|global default group during "+
+			"partition-map drift")
+
+	// A non-partitioned key is NOT recognised → must still fall
+	// through to the engine. This pins that fail-closed only fires
+	// for the recognised-but-unresolved case.
+	gid, ok := router.ResolveGroup([]byte("legacy-key"))
+	require.True(t, ok)
+	require.Equal(t, uint64(1), gid,
+		"non-recognised key must fall through to engine default")
+}
+
 // TestShardRouter_GetUsesResolver pins that the resolver-first path
 // applies to Get as well as Commit/Abort. A regression that fixed
 // only Commit's path would silently route reads through the engine
@@ -227,6 +282,14 @@ func (r *recordingResolver) ResolveGroup(key []byte) (uint64, bool) {
 	return 0, false
 }
 
+// RecognisesPartitionedKey returns false unconditionally — these
+// router-side tests use the resolver-wins / fall-through paths
+// only; the recognised-but-unresolved path is exercised separately
+// via fakePartitionResolver.recognisedPrefix.
+func (r *recordingResolver) RecognisesPartitionedKey([]byte) bool {
+	return false
+}
+
 func (r *recordingResolver) seenKeys() [][]byte {
 	r.mu.Lock()
 	defer r.mu.Unlock()

kv/sharded_coordinator_partition_test.go

@@ -13,10 +13,16 @@ import (
 // stubResolver routes any key in claim to the named group; everything
 // else returns (0, false). Used by the coordinator-level resolver
 // regression tests so they don't depend on the adapter package.
+//
+// recognisedPrefix simulates the "recognised but unresolved" path
+// (codex round-2 P1 on PR #715) — keys with this prefix that miss
+// claim cause RecognisesPartitionedKey to return true so the
+// router fails closed instead of falling through to the engine.
 type stubResolver struct {
-	mu    sync.Mutex
-	claim map[string]uint64
-	calls [][]byte
+	mu               sync.Mutex
+	claim            map[string]uint64
+	recognisedPrefix []byte
+	calls            [][]byte
 }
 
 func (s *stubResolver) ResolveGroup(key []byte) (uint64, bool) {
@@ -29,6 +35,14 @@ func (s *stubResolver) ResolveGroup(key []byte) (uint64, bool) {
 	return 0, false
 }
 
+func (s *stubResolver) RecognisesPartitionedKey(key []byte) bool {
+	if len(s.recognisedPrefix) == 0 {
+		return false
+	}
+	return len(key) >= len(s.recognisedPrefix) &&
+		string(key[:len(s.recognisedPrefix)]) == string(s.recognisedPrefix)
+}
+
 func (s *stubResolver) callKeys() [][]byte {
 	s.mu.Lock()
 	defer s.mu.Unlock()