fix(kv): Composed-1 verifyOwnerFromSnapshot must routeKey-normalize before OwnerOf (issue #930) (#932)

**Closes #930.**

`verifyOwnerFromSnapshot` was comparing the **raw mutation key** against
the engine's **routing-key ranges**, which sit in a completely different
lexicographical band. This caused every adapter write (DynamoDB, SQS,
S3, Redis internal) that crossed into a non-default Raft group to be
rejected with a spurious `ErrComposed1Violation`.

## 再現 (Issue #930)

1. Multi-group cluster を `--shardRanges ":<T3_KEY>=1,<T3_KEY>:=2"` で起動
2. M5a multi-table workload (PRs #911 / #916 / #924 / #925) を実行
3. Group 2 に routing されるテーブルへの全 CreateTable / Put / Get が以下の FSM error
で失敗:

```
observed-version v=1: key "!ddb|meta|table|amVwc2VuX2FwcGVuZF90Mw"
owned by group 1 (found=true); this FSM serves group 2:
composed-1: route ownership shifted; retry on new owning group
```

4. Workers が全 txn で `DynamoDB ResourceNotFoundException`、 Elle は
`:empty-transaction-graph` を report

## Root cause

`OwnerOf(rawKey)` の挙動:
- `routes[].Start` と `routes[].End` は **routing key**
(`!ddb|route|table|...`) の namespace
- でも `mut.Key` は **raw user key** (`!ddb|meta|table|...`)
- `!ddb|meta|...` (ASCII `m`=109) と `!ddb|route|...` (ASCII `r`=114) で 5
文字目がズレる
- 結果: 全 `!ddb|meta|...` キーが lexicographically `T3_KEY=!ddb|route|...`
より小さく、 group 1 へ "fall" させてしまう
- 実際の routing (`engine.GetRoute(routeKey(rawKey))`) は正しく group 2 を返すので、
Dispatch と verify で不整合発生

## Fix

`snap.OwnerOf(routeKey(mut.Key))` で routing-key namespace で比較するよう
normalize。 `ShardRouter.ResolveGroup` と `ShardStore.GetAt` が
`engine.GetRoute(routeKey(rawKey))` するのと同じ semantic。

## E2E 検証

修正後の `./scripts/run-jepsen-m5-local.sh` 実行結果:

```
results.edn: {:valid? true}
```

Workers が txn を正常完了、 read で先行 append を全て確認、 修正前の
`ResourceNotFoundException` は完全に解消。

## Caller audit (loop directive)

`verifyOwnerFromSnapshot` は `verifyComposed1` の 2 箇所から呼ばれる:
- observed-version check (line 618)
- current-version check (line 626)

両方とも同じ mutation list を受けるので、 normalization は均等に適用される。 unexported なので外部
caller なし。

## Test plan

- [x] `./scripts/run-jepsen-m5-local.sh` — `{:valid? true}`
- [ ] FSM レベルのユニットテスト追加 (follow-up — multi-group FSM verification の test
infrastructure 整備とセット)
- [x] `go build ./...` — 0 issues

## 関連 PR

- #926 (`fix/m5a-local-host`): `--host 127.0.0.1` + `--shardRanges`
default coverage — これも routing 問題だが、 PR #926 が解決するのは `Dispatch` 失敗パス。 本
PR は `verifyComposed1` の routing-key normalization。 両方が必要。
- #911 / #916 / #924 / #925: Composed-1 M5a 実装 (全部マージ済)
- 親 design doc:
`docs/design/2026_05_29_partial_composed1_cross_group_commit_guard.md`


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **Bug Fixes**
* Fixed a route ownership verification issue that was incorrectly
flagging certain adapter write operations as violations. The system now
properly handles key comparisons during ownership checks, preventing
spurious errors and improving system reliability.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: bootjp

kv/fsm.go

@@ -695,11 +695,16 @@ func (f *kvFSM) verifyOwnerFromSnapshot(mutations []*pb.Mutation, snap RouteSnap
 		if isTxnInternalKey(mut.Key) {
 			continue
 		}
-		owner, found := snap.OwnerOf(mut.Key)
+		// routeKey-normalize before OwnerOf so the gate routes the
+		// same way as ShardRouter.ResolveGroup — raw adapter keys
+		// and route catalog ranges live in different lex bands
+		// (issue #930).
+		rKey := routeKey(mut.Key)
+		owner, found := snap.OwnerOf(rKey)
 		if !found || owner != f.shardGroupID {
 			return errors.Wrapf(ErrComposed1Violation,
-				"%s-version v=%d: key %q owned by group %d (found=%v); this FSM serves group %d",
-				phase, snapVer, mut.Key, owner, found, f.shardGroupID)
+				"%s-version v=%d: key %q (routeKey %q) owned by group %d (found=%v); this FSM serves group %d",
+				phase, snapVer, mut.Key, rKey, owner, found, f.shardGroupID)
 		}
 	}
 	return nil

kv/fsm_composed1_test.go

@@ -321,3 +321,43 @@ func TestVerifyComposed1_AbortPhaseSkipsGate(t *testing.T) {
 	require.NoError(t, fsm.verifyComposed1(abortReq),
 		"ABORT requests must bypass the Composed-1 gate even with a non-zero ObservedRouteVersion pinned, so a route shift between PREPARE and ABORT cannot strand the txn's intent locks past lock-resolver TTL")
 }
+
+// TestVerifyComposed1_DynamoMetaKeyNormalizedToRouteKeyBeforeOwnerOf
+// is the regression for issue #930. verifyOwnerFromSnapshot was
+// passing the raw adapter mutation key (e.g. "!ddb|meta|table|<seg>")
+// to OwnerOf, but the route catalog's ranges are keyed by routing
+// keys ("!ddb|route|table|<seg>"). "meta" (ASCII 'm'=109) sorts below
+// "route" ('r'=114), so every "!ddb|meta|..." key fell into the
+// empty-prefix range — the wrong group — and every cross-group
+// DynamoDB write was rejected with ErrComposed1Violation. The fix
+// normalizes the key via routeKey() before OwnerOf, matching what
+// ShardRouter.ResolveGroup and ShardStore.GetAt already do on the
+// engine side.
+//
+// Without the fix this test fails: OwnerOf returns group 1 for the
+// raw "!ddb|meta|table|anyseg" key (which is < the "!ddb|route|..."
+// boundary), so the FSM serving group 2 raises ErrComposed1Violation.
+// With routeKey() normalization the raw key maps to
+// "!ddb|route|table|anyseg" which falls into group 2's range and the
+// gate passes.
+func TestVerifyComposed1_DynamoMetaKeyNormalizedToRouteKeyBeforeOwnerOf(t *testing.T) {
+	t.Parallel()
+
+	tableSegment := []byte("anyseg")
+	routeBoundary := append([]byte("!ddb|route|table|"), tableSegment...)
+	rawMetaKey := append([]byte("!ddb|meta|table|"), tableSegment...)
+
+	e := distribution.NewEngine()
+	applyComposed1Snapshot(t, e, 1, []distribution.RouteDescriptor{
+		// group 1 owns the empty-prefix half [<empty>, routeBoundary).
+		// Without routeKey() normalization, rawMetaKey would fall in here.
+		{RouteID: 100, Start: []byte(""), End: routeBoundary, GroupID: 1, State: distribution.RouteStateActive},
+		// group 2 owns [routeBoundary, +inf). routeKey(rawMetaKey)
+		// equals routeBoundary, so the normalized key lands here.
+		{RouteID: 101, Start: routeBoundary, End: nil, GroupID: 2, State: distribution.RouteStateActive},
+	})
+	fsm := newComposed1FSM(t, e, 2) // this FSM serves group 2
+
+	require.NoError(t, fsm.verifyComposed1(commitTxnRequest(1, string(rawMetaKey))),
+		"issue #930: verifyOwnerFromSnapshot must routeKey-normalize raw adapter keys before OwnerOf, so the gate routes the same way as ShardRouter.ResolveGroup; without normalization the raw \"!ddb|meta|...\" key sorts below the \"!ddb|route|...\" range and is falsely owned by group 1")
+}

kv/sharded_coordinator_composed1_retry_test.go

@@ -758,8 +758,8 @@ func (r *stubPartitionResolver) RecognisesPartitionedKey(key []byte) bool {
 //   - Dispatch-entry auto-pin stamps ObservedRouteVersion =
 //     engine.Version() on the request.
 //   - FSM at gid R applies; verifyComposed1 calls route-history
-//     OwnerOf on the raw mutation key against the engine's
-//     snapshot.
+//     OwnerOf on the routeKey()-normalized mutation key against
+//     the engine's snapshot (issue #930 fix).
 //   - Engine's snapshot says gid B (default byte-range owner),
 //     not gid R.
 //   - Mismatch → ErrComposed1Violation.