From 69f2d932d03dfb45a1cd535418f8bbb076d8a43c Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Thu, 22 Oct 2026 23:18:53 +0900
Subject: [PATCH 1/5] test: verify DynamoDB snapshot seals with real Raft

---
 .github/workflows/jepsen-test.yml             |  25 +-
 adapter/dynamodb_item_write.go                |   2 +-
 .../dynamodb_snapshot_seal_benchmark_test.go  | 102 +++++
 adapter/dynamodb_snapshot_seal_raft_test.go   | 366 ++++++++++++++++++
 adapter/dynamodb_transact.go                  |   1 +
 adapter/dynamodb_write_error.go               |  29 ++
 adapter/dynamodb_write_error_test.go          |  42 ++
 ...26_10_11_partial_dynamodb_snapshot_seal.md |  59 ++-
 .../src/elastickv/dynamodb_seal_workload.clj  |  90 +++++
 .../elastickv/dynamodb_seal_workload_test.clj |  34 ++
 main.go                                       |   5 +
 main_dynamodb_follower_test.go                |  25 ++
 12 files changed, 771 insertions(+), 9 deletions(-)
 create mode 100644 adapter/dynamodb_snapshot_seal_benchmark_test.go
 create mode 100644 adapter/dynamodb_snapshot_seal_raft_test.go
 create mode 100644 adapter/dynamodb_write_error.go
 create mode 100644 adapter/dynamodb_write_error_test.go
 create mode 100644 jepsen/src/elastickv/dynamodb_seal_workload.clj
 create mode 100644 jepsen/test/elastickv/dynamodb_seal_workload_test.clj
 create mode 100644 main_dynamodb_follower_test.go

diff --git a/.github/workflows/jepsen-test.yml b/.github/workflows/jepsen-test.yml
index 2061f6e46..a002ddf04 100644
--- a/.github/workflows/jepsen-test.yml
+++ b/.github/workflows/jepsen-test.yml
@@ -64,6 +64,7 @@ jobs:
           go mod download
           go build -o /tmp/elastickv-bin .
           go build -o /tmp/elastickv-admin-bin ./cmd/elastickv-admin
+          go build -o /tmp/elastickv-raftadmin-bin ./cmd/raftadmin
       - name: Warm Leiningen Maven cache
         working-directory: jepsen
         run: |
@@ -90,7 +91,8 @@ jobs:
           BOOTSTRAP_MEMBERS="n1=127.0.0.1:50051,n2=127.0.0.1:50052,n3=127.0.0.1:50053"
           RAFT_REDIS_MAP="127.0.0.1:50051=127.0.0.1:63791,127.0.0.1:50052=127.0.0.1:63792,127.0.0.1:50053=127.0.0.1:63793"
           RAFT_S3_MAP="127.0.0.1:50051=127.0.0.1:63901,127.0.0.1:50052=127.0.0.1:63902,127.0.0.1:50053=127.0.0.1:63903"
-          RAFT_DYNAMO_MAP="127.0.0.1:50051=127.0.0.1:63801,127.0.0.1:50052=127.0.0.1:63802,127.0.0.1:50053=127.0.0.1:63803"
+          # No raftDynamoMap: DynamoDB workloads must exercise follower gRPC
+          # forwarding and leader-issued seals instead of HTTP leader proxying.
           RAFT_SQS_MAP="127.0.0.1:50051=127.0.0.1:63501,127.0.0.1:50052=127.0.0.1:63502,127.0.0.1:50053=127.0.0.1:63503"
 
           # Encrypted leg: one random owner-only file KEK shared by the
@@ -132,7 +134,6 @@ jobs:
               --raftBootstrapMembers "$BOOTSTRAP_MEMBERS" \
               --raftRedisMap "$RAFT_REDIS_MAP" \
               --raftS3Map "$RAFT_S3_MAP" \
-              --raftDynamoMap "$RAFT_DYNAMO_MAP" \
               --raftSqsMap "$RAFT_SQS_MAP" \
               "${ENC_ARGS[@]}" \
               > "/tmp/elastickv-demo-n${node}.log" 2>&1 &
@@ -183,6 +184,26 @@ jobs:
         timeout-minutes: 3
         run: |
           timeout 120 ~/lein run -m elastickv.dynamodb-workload --local --time-limit 5 --rate 5 --concurrency 5 --dynamo-ports 63801,63802,63803 --host 127.0.0.1
+      - name: Run DynamoDB seal workload with recorded leader changes
+        working-directory: jepsen
+        run: |
+          printf 'run_sha=%s\n' "$GITHUB_SHA" > /tmp/dynamodb-seal-source.txt
+          timeout 150 ~/lein run -m elastickv.dynamodb-seal-workload --local \
+            --time-limit 15 --rate 30 --concurrency 9 --max-txn-length 4 \
+            --dynamo-ports 63801,63802,63803 --host 127.0.0.1 \
+            --raftadmin-bin /tmp/elastickv-raftadmin-bin \
+            --raft-addresses 127.0.0.1:50051,127.0.0.1:50052,127.0.0.1:50053 \
+            --leader-change-interval 2
+      - name: Upload DynamoDB seal acceptance history
+        if: always()
+        uses: actions/upload-artifact@v7
+        with:
+          name: dynamodb-seal-encryption-${{ matrix.encryption }}
+          path: |
+            jepsen/store/elastickv-dynamodb-snapshot-seal/
+            /tmp/dynamodb-seal-source.txt
+          retention-days: 7
+          if-no-files-found: ignore
       - name: Run DynamoDB per-type Jepsen workloads against elastickv
         working-directory: jepsen
         timeout-minutes: 10
diff --git a/adapter/dynamodb_item_write.go b/adapter/dynamodb_item_write.go
index 761545a80..1354ff184 100644
--- a/adapter/dynamodb_item_write.go
+++ b/adapter/dynamodb_item_write.go
@@ -485,7 +485,7 @@ func (d *DynamoDBServer) commitItemWrite(ctx context.Context, readTimestamp kv.R
 	dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
 	_, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req)
 	if err != nil {
-		return errors.WithStack(err)
+		return errors.WithStack(normalizeDynamoWriteDispatchError(err))
 	}
 	return nil
 }
diff --git a/adapter/dynamodb_snapshot_seal_benchmark_test.go b/adapter/dynamodb_snapshot_seal_benchmark_test.go
new file mode 100644
index 000000000..9b0f4d752
--- /dev/null
+++ b/adapter/dynamodb_snapshot_seal_benchmark_test.go
@@ -0,0 +1,102 @@
+package adapter
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"testing"
+
+	"github.com/bootjp/elastickv/kv"
+	"github.com/bootjp/elastickv/store"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+)
+
+type sealBenchmarkCoordinator struct {
+	kv.Coordinator
+	seals     atomic.Uint64
+	conflicts atomic.Uint64
+}
+
+func (c *sealBenchmarkCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
+	if isDynamoSnapshotSealRequest(req) {
+		c.seals.Add(1)
+	}
+	resp, err := c.Coordinator.Dispatch(ctx, req)
+	if errors.Is(err, store.ErrWriteConflict) {
+		c.conflicts.Add(1)
+	}
+	return resp, err
+}
+
+// A single durable Raft voter isolates the extra seal/OCC cost. This is not a
+// networked quorum or a production throughput claim. Separate frontends avoid
+// serializing the workload with a shared server's item-lock stripes.
+func BenchmarkDynamoDBSnapshotSealRealRaft(b *testing.B) {
+	for _, mode := range []string{"independent", "hot_item", "collision"} {
+		b.Run(mode, func(b *testing.B) {
+			ctx := context.Background()
+			st := store.NewMVCCStore()
+			defer st.Close()
+			server, _, stop := sealRaftRestartableServer(b, b.TempDir(), st)
+			defer stop()
+			schema := dedupItemTable()
+			require.NoError(b, server.createTableWithRetry(ctx, schema.TableName, schema))
+			keys := sealBenchmarkKeys(b, schema, mode)
+			for _, key := range keys {
+				_, err := server.putItemWithRetry(ctx, putItemInput{
+					TableName: schema.TableName, Item: map[string]attributeValue{"pk": newStringAttributeValue(key)},
+				})
+				require.NoError(b, err)
+			}
+			coord := &sealBenchmarkCoordinator{Coordinator: server.coordinator}
+			var worker atomic.Uint64
+			b.ReportAllocs()
+			b.ResetTimer()
+			b.RunParallel(func(pb *testing.PB) {
+				local := NewDynamoDBServer(nil, st, coord)
+				key := keys[(worker.Add(1)-1)%uint64(len(keys))]
+				in := putItemInput{TableName: schema.TableName, Item: map[string]attributeValue{
+					"pk": newStringAttributeValue(key), "value": newStringAttributeValue("constant-size replacement"),
+				}}
+				for pb.Next() {
+					if _, err := local.putItemWithRetry(ctx, in); err != nil {
+						b.Error(err)
+						return
+					}
+				}
+			})
+			b.StopTimer()
+			b.ReportMetric(float64(coord.seals.Load())/float64(b.N), "seals/op")
+			b.ReportMetric(float64(coord.conflicts.Load())/float64(b.N), "conflicts/op")
+		})
+	}
+}
+
+func sealBenchmarkKeys(b *testing.B, schema *dynamoTableSchema, mode string) []string {
+	b.Helper()
+	if mode == "hot_item" {
+		return []string{"hot"}
+	}
+	const count = 4
+	buckets := make(map[string][]string)
+	independent := make([]string, 0, count)
+	for i := range 1 << 20 {
+		key := fmt.Sprintf("seal-bench-%d", i)
+		item, err := schema.itemKeyFromAttributes(map[string]attributeValue{"pk": newStringAttributeValue(key)})
+		require.NoError(b, err)
+		seal := string(dynamoItemSnapshotSealKey(schema.TableName, schema.Generation, item))
+		if mode == "independent" && len(buckets[seal]) == 0 {
+			independent = append(independent, key)
+			if len(independent) == count {
+				return independent
+			}
+		}
+		buckets[seal] = append(buckets[seal], key)
+		if mode == "collision" && len(buckets[seal]) == count {
+			return buckets[seal]
+		}
+	}
+	b.Fatal("could not construct seal bucket fixture")
+	return nil
+}
diff --git a/adapter/dynamodb_snapshot_seal_raft_test.go b/adapter/dynamodb_snapshot_seal_raft_test.go
new file mode 100644
index 000000000..814c48dea
--- /dev/null
+++ b/adapter/dynamodb_snapshot_seal_raft_test.go
@@ -0,0 +1,366 @@
+package adapter
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/bootjp/elastickv/distribution"
+	"github.com/bootjp/elastickv/internal/raftengine"
+	etcdraftengine "github.com/bootjp/elastickv/internal/raftengine/etcd"
+	"github.com/bootjp/elastickv/kv"
+	"github.com/bootjp/elastickv/store"
+	json "github.com/goccy/go-json"
+	"github.com/stretchr/testify/require"
+)
+
+// Pause after the adapter allocates C, but before the real coordinator proposes
+// it. No store/FSM/transport is mocked: the later seal must reject this plan.
+type delayedSealRaftCoordinator struct {
+	kv.Coordinator
+	blocked   atomic.Bool
+	entered   chan uint64
+	release   chan struct{}
+	result    chan error
+	unstamped bool
+}
+
+func (c *delayedSealRaftCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
+	if req == nil || !req.IsTxn || (req.CommitTS == 0 && !c.unstamped) || isDynamoSnapshotSealRequest(req) || !c.blocked.CompareAndSwap(false, true) {
+		return c.Coordinator.Dispatch(ctx, req)
+	}
+	stamp := req.CommitTS
+	if stamp == 0 {
+		stamp = req.StartTS
+	}
+	c.entered <- stamp
+	select {
+	case <-c.release:
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+	resp, err := c.Coordinator.Dispatch(ctx, req)
+	c.result <- err
+	return resp, err
+}
+
+func newDelayedSealRaftServer(t *testing.T, server *DynamoDBServer, unstamped bool) (*DynamoDBServer, *delayedSealRaftCoordinator, func()) {
+	t.Helper()
+	coord := &delayedSealRaftCoordinator{
+		Coordinator: server.coordinator,
+		entered:     make(chan uint64, 1),
+		release:     make(chan struct{}),
+		result:      make(chan error, 1),
+		unstamped:   unstamped,
+	}
+	var once sync.Once
+	release := func() { once.Do(func() { close(coord.release) }) }
+	t.Cleanup(release)
+	return NewDynamoDBServer(nil, server.store, coord), coord, release
+}
+
+func sealRaftSeed(t *testing.T, ctx context.Context, server *DynamoDBServer, schema *dynamoTableSchema) {
+	t.Helper()
+	require.NoError(t, server.createTableWithRetry(ctx, schema.TableName, schema))
+	_, err := server.putItemWithRetry(ctx, putItemInput{TableName: schema.TableName, Item: map[string]attributeValue{
+		"pk": newStringAttributeValue(dedupTestKey), "val": numberListAttr("1"),
+	}})
+	require.NoError(t, err)
+}
+
+func sealRaftAppend(table, value string) updateItemInput {
+	in := appendListInput()
+	in.TableName = table
+	in.ExpressionAttributeValues[":val"] = numberListAttr(value)
+	return in
+}
+
+func sealRaftTxnUpdate(table, value string) transactWriteItem {
+	in := sealRaftAppend(table, value)
+	return transactWriteItem{Update: &transactUpdateInput{
+		TableName: in.TableName, Key: in.Key, UpdateExpression: in.UpdateExpression,
+		ExpressionAttributeNames: in.ExpressionAttributeNames, ExpressionAttributeValues: in.ExpressionAttributeValues,
+	}}
+}
+
+func sealRaftDelayedAppend(ctx context.Context, server *DynamoDBServer, table string) <-chan error {
+	done := make(chan error, 1)
+	go func() {
+		_, err := server.updateItemWithRetry(ctx, sealRaftAppend(table, "2"))
+		done <- err
+	}()
+	return done
+}
+
+func sealRaftReceive[T any](t *testing.T, ctx context.Context, ch <-chan T) T {
+	t.Helper()
+	select {
+	case value := <-ch:
+		return value
+	case <-ctx.Done():
+		t.Fatalf("waiting for delayed Raft write: %v", ctx.Err())
+		var zero T
+		return zero
+	}
+}
+
+func sealRaftHTTP(t *testing.T, ctx context.Context, address, action string, input any) {
+	t.Helper()
+	require.NoError(t, sealRaftHTTPCall(ctx, address, action, input))
+}
+
+func sealRaftHTTPCall(ctx context.Context, address, action string, input any) error {
+	body, err := json.Marshal(input)
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://"+address+"/", bytes.NewReader(body))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("X-Amz-Target", "DynamoDB_20120810."+action)
+	req.Header.Set("Content-Type", "application/x-amz-json-1.0")
+	client := &http.Client{Timeout: 10 * time.Second}
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	response, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("DynamoDB HTTP %d: %s", resp.StatusCode, response)
+	}
+	return nil
+}
+
+func TestDynamoDBSnapshotSealRealRaftLeaderChangeFollowerIngress(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	nodes, _, _ := createNode(t, 3)
+	defer shutdown(nodes)
+	schema := dedupItemTable()
+	leader := nodes[0].dynamoServer
+	sealRaftSeed(t, ctx, leader, schema)
+	delayed, coord, release := newDelayedSealRaftServer(t, leader, false)
+	defer release()
+	done := sealRaftDelayedAppend(ctx, delayed, schema.TableName)
+	commitTS := sealRaftReceive(t, ctx, coord.entered)
+	sealRaftAdvanceWatermark(t, ctx, leader.coordinator, commitTS)
+
+	admin, ok := nodes[0].engine.(raftengine.Admin)
+	require.True(t, ok)
+	require.NoError(t, admin.TransferLeadershipToServer(ctx, "1", nodes[1].raftAddress))
+	require.Eventually(t, func() bool {
+		return nodes[1].engine.State() == raftengine.StateLeader &&
+			nodes[0].engine.State() == raftengine.StateFollower && nodes[2].engine.State() == raftengine.StateFollower &&
+			nodes[0].engine.Leader().Address == nodes[1].raftAddress &&
+			nodes[2].engine.Leader().Address == nodes[1].raftAddress
+	}, 5*time.Second, 10*time.Millisecond)
+	follower := nodes[2].dynamoServer
+	require.Empty(t, follower.leaderDynamo, "HTTP forwarding would hide the follower timestamp path")
+	second, secondCoord, releaseSecond := newDelayedSealRaftServer(t, follower, true)
+	defer releaseSecond()
+	httpServer := httptest.NewServer(http.HandlerFunc(second.handle))
+	defer func() {
+		releaseSecond()
+		cancel()
+		httpServer.Close()
+	}()
+	txnDone := make(chan error, 1)
+	go func() {
+		txnDone <- sealRaftHTTPCall(ctx, strings.TrimPrefix(httpServer.URL, "http://"), "TransactWriteItems", transactWriteItemsInput{
+			TransactItems: []transactWriteItem{sealRaftTxnUpdate(schema.TableName, "3")},
+		})
+	}()
+	var sealed uint64
+	select {
+	case sealed = <-secondCoord.entered:
+	case err := <-txnDone:
+		t.Fatalf("transaction finished before the planned-write barrier: %v", err)
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	}
+	require.Greater(t, sealed, commitTS)
+	release()
+	require.ErrorContains(t, sealRaftReceive(t, ctx, coord.result), dynamoSealPrefix, "only the seal changed since the first read")
+	require.NoError(t, sealRaftReceive(t, ctx, done))
+	releaseSecond()
+	require.NoError(t, sealRaftReceive(t, ctx, txnDone))
+	require.Equal(t, []string{"1", "2", "3"}, readListValues(t, nodes[1].dynamoServer, schema))
+	// Also exercise the follower's single-item legacy timestamp allocation.
+	sealRaftHTTP(t, ctx, nodes[2].dynamoAddress, "UpdateItem", sealRaftAppend(schema.TableName, "4"))
+	require.Equal(t, []string{"1", "2", "3", "4"}, readListValues(t, nodes[1].dynamoServer, schema))
+	t.Logf("delayed C=%d, successor seal R=%d, old write rejected before transaction commit", commitTS, sealed)
+}
+
+func sealRaftAdvanceWatermark(t *testing.T, ctx context.Context, coord kv.Coordinator, pending uint64) {
+	t.Helper()
+	ts, err := kv.NextTimestampAfterThrough(ctx, coord, pending, "advance unrelated watermark")
+	require.NoError(t, err)
+	require.Greater(t, ts, pending)
+	_, err = coord.Dispatch(ctx, &kv.OperationGroup[kv.OP]{
+		CommitTS: ts,
+		Elems:    []*kv.Elem[kv.OP]{{Op: kv.Put, Key: []byte("watermark"), Value: []byte("x")}},
+	})
+	require.NoError(t, err)
+}
+
+func TestDynamoDBSnapshotSealRealRaftTwoDataGroups(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+	left, right := dedupItemTable(), dedupItemTable()
+	left.TableName, right.TableName = "alpha", "beta"
+	split := kv.RouteKey(dynamoTableMetaKey(right.TableName))
+	require.Negative(t, bytes.Compare(kv.RouteKey(dynamoTableMetaKey(left.TableName)), split))
+	routes := distribution.NewEngine()
+	routes.UpdateRoute(nil, split, 1)
+	routes.UpdateRoute(split, nil, 2)
+	hlc := kv.NewHLC()
+	st1, st2 := store.NewMVCCStore(), store.NewMVCCStore()
+	eng1, stop1 := newSingleRaftForDistributionE2E(t, "seal-group-1", kv.NewKvFSMWithHLC(st1, hlc))
+	defer stop1()
+	eng2, stop2 := newSingleRaftForDistributionE2E(t, "seal-group-2", kv.NewKvFSMWithHLC(st2, hlc))
+	defer stop2()
+	groups := map[uint64]*kv.ShardGroup{
+		1: {Engine: eng1, Store: st1, Txn: kv.NewLeaderProxyWithEngine(eng1)},
+		2: {Engine: eng2, Store: st2, Txn: kv.NewLeaderProxyWithEngine(eng2)},
+	}
+	st := kv.NewShardStore(routes, groups)
+	defer func() {
+		stop1()
+		stop2()
+		require.NoError(t, st.Close())
+	}()
+	coordinator := kv.NewShardedCoordinator(routes, groups, 1, hlc, st)
+	server := NewDynamoDBServer(nil, st, coordinator)
+	sealRaftSeed(t, ctx, server, left)
+	sealRaftSeed(t, ctx, server, right)
+	delayed, coord, release := newDelayedSealRaftServer(t, server, false)
+	defer release()
+	done := sealRaftDelayedAppend(ctx, delayed, right.TableName)
+	commitTS := sealRaftReceive(t, ctx, coord.entered)
+	sealRaftAdvanceWatermark(t, ctx, coordinator, commitTS)
+
+	// This independent frontend has no shared per-item locks with the writer.
+	second, secondCoord, releaseSecond := newDelayedSealRaftServer(t, server, true)
+	defer releaseSecond()
+	input := transactWriteItemsInput{TransactItems: []transactWriteItem{
+		sealRaftTxnUpdate(left.TableName, "3"), sealRaftTxnUpdate(right.TableName, "3"),
+	}}
+	body, err := json.Marshal(input)
+	require.NoError(t, err)
+	recorder := httptest.NewRecorder()
+	txnDone := make(chan struct{})
+	go func() {
+		second.transactWriteItems(recorder, httptest.NewRequestWithContext(ctx, http.MethodPost, "/", bytes.NewReader(body)))
+		close(txnDone)
+	}()
+	sealed := sealRaftReceive(t, ctx, secondCoord.entered)
+	require.Greater(t, sealed, commitTS)
+	key := map[string]attributeValue{"pk": newStringAttributeValue(dedupTestKey)}
+	leftKey, err := left.itemKeyFromAttributes(key)
+	require.NoError(t, err)
+	rightKey, err := right.itemKeyFromAttributes(key)
+	require.NoError(t, err)
+	seal1, seal2 := dynamoItemSnapshotSealKey(left.TableName, 1, leftKey), dynamoItemSnapshotSealKey(right.TableName, 1, rightKey)
+	ts1, found, err := st1.LatestCommitTS(ctx, seal1)
+	require.NoError(t, err)
+	require.True(t, found)
+	ts2, found, err := st2.LatestCommitTS(ctx, seal2)
+	require.NoError(t, err)
+	require.True(t, found)
+	require.Greater(t, ts1, commitTS)
+	require.Equal(t, ts1, ts2, "both real Raft groups must seal at the same R")
+	release()
+	require.ErrorIs(t, sealRaftReceive(t, ctx, coord.result), store.ErrWriteConflict)
+	require.NoError(t, sealRaftReceive(t, ctx, done))
+	releaseSecond()
+	sealRaftReceive(t, ctx, txnDone)
+	require.Equal(t, http.StatusOK, recorder.Code, "%s", recorder.Body.String())
+	require.Equal(t, []string{"1", "3"}, readListValues(t, second, left))
+	require.Equal(t, []string{"1", "2", "3"}, readListValues(t, second, right))
+	t.Logf("two data groups: delayed C=%d, shared R=%d", commitTS, ts1)
+}
+
+func TestDynamoDBSnapshotSealRealRaftSnapshotRestart(t *testing.T) {
+	t.Setenv("ELASTICKV_RAFT_SNAPSHOT_COUNT", "1")
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+	dir := t.TempDir()
+	st := store.NewMVCCStore()
+	server, engine, stop := sealRaftRestartableServer(t, dir, st)
+	schema := dedupItemTable()
+	sealRaftSeed(t, ctx, server, schema)
+	key, err := schema.itemKeyFromAttributes(map[string]attributeValue{"pk": newStringAttributeValue(dedupTestKey)})
+	require.NoError(t, err)
+	seal := dynamoItemSnapshotSealKey(schema.TableName, 1, key)
+	before, found, err := st.LatestCommitTS(ctx, seal)
+	require.NoError(t, err)
+	require.True(t, found)
+	stamp, err := server.beginTxnWriteTimestamp(ctx, "restart seal", [][]byte{seal})
+	require.NoError(t, err)
+	sealed := stamp.Timestamp()
+	require.Greater(t, sealed, before)
+	index := engine.Status().AppliedIndex
+	require.Eventually(t, func() bool { return engine.Status().LastSnapshotIndex >= index }, 5*time.Second, 10*time.Millisecond)
+	stop()
+	require.NoError(t, st.Close())
+
+	// Reopen the same Raft directory with a genuinely empty state machine.
+	// The item and seal below must come from its durable snapshot/WAL, not RAM.
+	restored := store.NewMVCCStore()
+	restarted, _, stopRestarted := sealRaftRestartableServer(t, dir, restored)
+	defer restored.Close()
+	defer stopRestarted()
+	sealTS, found, err := restored.LatestCommitTS(ctx, seal)
+	require.NoError(t, err)
+	require.True(t, found)
+	require.Equal(t, sealed, sealTS)
+	require.Equal(t, []string{"1"}, readListValues(t, restarted, schema))
+	_, err = restarted.coordinator.Dispatch(ctx, &kv.OperationGroup[kv.OP]{
+		IsTxn: true, StartTS: before, ReadKeys: [][]byte{seal},
+		Elems: []*kv.Elem[kv.OP]{{Op: kv.Put, Key: key, Value: []byte("stale plan")}},
+	})
+	require.ErrorIs(t, err, store.ErrWriteConflict)
+	_, err = restarted.updateItemWithRetry(ctx, sealRaftAppend(schema.TableName, "2"))
+	require.NoError(t, err)
+	require.Equal(t, []string{"1", "2"}, readListValues(t, restarted, schema))
+}
+
+func sealRaftRestartableServer(t testing.TB, dir string, st store.MVCCStore) (*DynamoDBServer, raftengine.Engine, func()) {
+	t.Helper()
+	hlc := kv.NewHLC()
+	factory := etcdraftengine.NewFactory(etcdraftengine.FactoryConfig{
+		TickInterval: 10 * time.Millisecond, HeartbeatTick: 1, ElectionTick: 10,
+		MaxSizePerMsg: 1 << 20, MaxInflightMsg: 256,
+	})
+	result, err := factory.Create(raftengine.FactoryConfig{
+		LocalID: "seal-restart", LocalAddress: "seal-restart", DataDir: dir, Bootstrap: true,
+		StateMachine: kv.NewKvFSMWithHLC(st, hlc),
+	})
+	require.NoError(t, err)
+	var once sync.Once
+	stop := func() {
+		once.Do(func() {
+			require.NoError(t, result.Engine.Close())
+			if result.Close != nil {
+				require.NoError(t, result.Close())
+			}
+		})
+	}
+	t.Cleanup(stop)
+	require.Eventually(t, func() bool { return result.Engine.State() == raftengine.StateLeader }, 5*time.Second, 10*time.Millisecond)
+	coord := kv.NewCoordinatorWithEngine(kv.NewTransactionWithProposer(result.Engine), result.Engine, kv.WithHLC(hlc))
+	return NewDynamoDBServer(nil, st, coord), result.Engine, stop
+}
diff --git a/adapter/dynamodb_transact.go b/adapter/dynamodb_transact.go
index ba371af6b..217437b2d 100644
--- a/adapter/dynamodb_transact.go
+++ b/adapter/dynamodb_transact.go
@@ -753,6 +753,7 @@ func (d *DynamoDBServer) runTransactWriteAttempt(
 	}
 	dispatchCtx := reqs.readTimestamp.WithDispatchVoucher(ctx)
 	if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, reqs.group); err != nil {
+		err = normalizeDynamoWriteDispatchError(err)
 		wrapped := errors.WithStack(err)
 		if !isRetryableTransactWriteError(err) {
 			return false, wrapped
diff --git a/adapter/dynamodb_write_error.go b/adapter/dynamodb_write_error.go
new file mode 100644
index 000000000..51ab9f3c4
--- /dev/null
+++ b/adapter/dynamodb_write_error.go
@@ -0,0 +1,29 @@
+package adapter
+
+import (
+	"strings"
+
+	"github.com/bootjp/elastickv/store"
+	"github.com/cockroachdb/errors"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// Forwarded OCC conflicts lose their Go error chain. Restore only the known
+// wire envelope so item retries retain the existing exact-timestamp dedup path.
+func normalizeDynamoWriteDispatchError(err error) error {
+	if err == nil || errors.Is(err, store.ErrWriteConflict) {
+		return err
+	}
+	type statusCarrier interface{ GRPCStatus() *status.Status }
+	var carrier statusCarrier
+	if !errors.As(err, &carrier) {
+		return err
+	}
+	st := carrier.GRPCStatus()
+	if st != nil && (st.Code() == codes.Unknown || st.Code() == codes.Aborted) &&
+		strings.HasPrefix(st.Message(), "key: ") && strings.HasSuffix(st.Message(), ": "+store.ErrWriteConflict.Error()) {
+		return errors.WithStack(errors.Mark(err, store.ErrWriteConflict))
+	}
+	return err
+}
diff --git a/adapter/dynamodb_write_error_test.go b/adapter/dynamodb_write_error_test.go
new file mode 100644
index 000000000..5de6e53a8
--- /dev/null
+++ b/adapter/dynamodb_write_error_test.go
@@ -0,0 +1,42 @@
+package adapter
+
+import (
+	"testing"
+
+	"github.com/bootjp/elastickv/store"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func TestNormalizeDynamoWriteDispatchError(t *testing.T) {
+	t.Parallel()
+	for _, tc := range []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"nil", nil, false},
+		{"local conflict", store.NewWriteConflictError([]byte("k")), true},
+		{"unknown conflict", status.Error(codes.Unknown, "key: k: write conflict"), true},
+		{"wrapped aborted conflict", errors.WithStack(status.Error(codes.Aborted, "key: k: write conflict")), true},
+		{"plain text", errors.New("key: k: write conflict"), false},
+		{"missing key envelope", status.Error(codes.Unknown, "write conflict"), false},
+		{"wrong code", status.Error(codes.Internal, "key: k: write conflict"), false},
+		{"embedded sentinel", status.Error(codes.Unknown, "key: write conflict: invalid value"), false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			normalized := normalizeDynamoWriteDispatchError(tc.err)
+			require.Equal(t, tc.want, errors.Is(normalized, store.ErrWriteConflict))
+			switch {
+			case tc.want:
+				require.True(t, shouldPreserveTransactWriteAttempt(normalized))
+			case tc.err != nil:
+				require.Same(t, tc.err, normalized)
+			default:
+				require.NoError(t, normalized)
+			}
+		})
+	}
+}
diff --git a/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md b/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
index 7b43e83ae..fcb3078e6 100644
--- a/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
+++ b/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
@@ -90,11 +90,58 @@ This is three-node smoke evidence, not evidence of the follower path with the HT
 [`elastickv-dynamodb-test`](../../jepsen/src/elastickv/dynamodb_workload.clj) disables the nemesis in local mode.
 The multi-table workload exists in [`dynamodb_multi_table_workload.clj`](../../jepsen/src/elastickv/dynamodb_multi_table_workload.clj), but that existence and its unit tests are not a recorded execution of the seal failure schedule across data groups.
 
-### Remaining acceptance work
+### Follow-up acceptance tests (2026-10-22)
 
-- Run the allocated/unproposed-write regression through real Raft with follower ingress and an election, retaining the history and final item assertion.
-- Repeat the delayed-write schedule with two touched data groups, and cover seal persistence through snapshot restore and process restart.
-- Exercise follower ingress with the HTTP leader map disabled and leader changes in Jepsen, retaining the exact revision, topology, nemesis configuration, histories, and checker results.
-- Measure the extra seal transaction under hot-item contention, digest collisions, and independent buckets. No seal-specific benchmark function or recorded contention measurement was found in the audited tracked files, and #1259 explicitly reports production throughput as unmeasured. Existing Redis benchmarks are not evidence for this requirement.
+The audit above describes main at `429f5ee7`; the following additions address its validation gaps without replacing #1259's seal protocol.
 
-These are validation and measurement gaps. This audit found the seal protocol and all item entry-point wiring already present; it does not add a second implementation or claim that the remaining acceptance work has passed.
+[`dynamodb_snapshot_seal_raft_test.go`](../../adapter/dynamodb_snapshot_seal_raft_test.go) uses the existing etcd/Raft test fixtures:
+
+- `TestDynamoDBSnapshotSealRealRaftLeaderChangeFollowerIngress` holds a write after allocating C, advances an unrelated key, transfers real three-voter Raft leadership, and sends HTTP to a follower with no HTTP leader map. It pauses the second transaction after sealing and planning, but before commit. The old write must fail on the **seal key**, before any second item write commits. Both retried appends must survive, in order, without duplicates. The test also writes through the follower's single-item path. The leader-change case passed ten consecutive race-enabled runs after waiting for every replica to recognize the new leader.
+- `TestDynamoDBSnapshotSealRealRaftTwoDataGroups` repeats the delayed-write schedule using two independent durable one-voter Raft groups, one per table route. It checks that the stored seals have the same R in the two different stores, rejects the old write, and asserts both tables' contents. This is not a two-group, three-replica-per-group topology.
+- `TestDynamoDBSnapshotSealRealRaftSnapshotRestart` forces a durable Raft snapshot, closes the engine and store, and opens the same Raft directory with an empty FSM. It checks the exact restored seal timestamp and item, rejects a stale plan against the restored seal, then accepts a fresh append. This fixture is a cold engine/FSM restart inside the test process, not an OS process-kill experiment.
+
+These tests exposed two runtime gaps, rather than a missing seal implementation:
+
+1. Forwarded write conflicts arrive as gRPC statuses without the local sentinel chain. [`normalizeDynamoWriteDispatchError`](../../adapter/dynamodb_write_error.go) recognizes only the known key-conflict envelope and restores the sentinel at DynamoDB's item/transaction dispatch boundaries. Existing dedup and recompute behavior remains in place. The shared retry predicates, including their SQS callers, are unchanged.
+2. [`buildLeaderDynamo`](../../main.go) previously synthesized a self-only HTTP map even when the flag was omitted. A follower could not resolve a remote leader in that map. An omitted DynamoDB map now selects the existing coordinator gRPC path; explicit maps retain the previous local-entry completion. [`main_dynamodb_follower_test.go`](../../main_dynamodb_follower_test.go) covers both modes and malformed configuration. Redis, S3, and SQS map builders are unchanged.
+
+All production callers of the changed functions were audited with `rg`: three `commitItemWrite` callers (legacy, first dedup attempt, reused dedup attempt), one `runTransactWriteAttempt` caller, and one `buildLeaderDynamo` caller in runtime configuration.
+
+### Follower-ingress Jepsen evidence
+
+[`dynamodb_seal_workload.clj`](../../jepsen/src/elastickv/dynamodb_seal_workload.clj) reuses the existing DynamoDB append workload and Jepsen generator/checker library. Its local nemesis invokes the repository's `raftadmin`, verifies that leadership actually moves to a different node, and records the endpoints in the history. A separate checker rejects zero verified transfers or any failed transfer; a successful append checker alone is insufficient.
+
+The local run `20261022T231204.299+0900` used three separate server processes on loopback, no `raftDynamoMap`, 15 seconds, rate 30, concurrency 9, maximum transaction length 4, and a two-second transfer interval. The checker output was:
+
+```clojure
+{:append {:valid? true},
+ :leader-changes {:valid? true, :verified-leader-changes 5, :failures []},
+ :valid? true}
+```
+
+The raw history, `test.jepsen`, checker results, and workload log remain in `jepsen/store/elastickv-dynamodb-snapshot-seal/20261022T231204.299+0900/` in the acceptance worktree. The external launcher retains the per-node logs and topology separately. This is short-run safety evidence, not a claim of production availability or a partition/kill campaign; indeterminate client results are permitted by the existing workload.
+
+The [`Jepsen CI workflow`](../../.github/workflows/jepsen-test.yml) now omits the HTTP leader map and runs this workload in both encryption legs. Its artifact retains the seal history and tested revision even on success. Those new CI executions must be checked separately from the local run.
+
+### Contention measurements
+
+[`BenchmarkDynamoDBSnapshotSealRealRaft`](../../adapter/dynamodb_snapshot_seal_benchmark_test.go) uses a durable single-voter Raft engine, four independent frontends, and constant-size PutItem replacements. The independent case uses four different buckets; the collision case searches for four different item keys mapping to one actual digest bucket. A hot-item case uses one item across frontends. Local item locks are not shared between frontends.
+
+Command: `go test ./adapter -run '^$' -bench '^BenchmarkDynamoDBSnapshotSealRealRaft$' -benchtime=100x -count=3 -cpu=4 -benchmem` on Darwin/arm64, Apple M1 Max.
+
+| Case | ns/op range (three samples) | OCC conflicts/op | Seal dispatches/op |
+| --- | ---: | ---: | ---: |
+| Independent buckets | 9,750,054-10,273,370 | 0 | 1.00 |
+| Hot item | 21,095,990-22,904,492 | 1.82-2.23 | 2.46-2.72 |
+| Digest collision | 21,698,244-23,050,694 | 1.97-2.01 | 2.59-2.62 |
+
+All measured operations completed. Collision/hot-item contention increased retries and seal work in this fixture. These measurements include adapter and local durable Raft costs, not network quorum latency, Pebble production compaction, or a before/after throughput delta against an unsafe seal-free implementation. Production throughput remains unmeasured.
+
+### Verification and remaining gates
+
+- The real-Raft tests, all DynamoDB and existing one-phase dedup tests passed with `-race`; the leader-change case also passed ten consecutive runs.
+- `TestBuildLeaderDynamoWithoutHTTPMap` and scoped main/adapter lint passed. Lint used the repository's Go 1.26.6 toolchain because the locally installed analyzer cannot consume Go 1.27 export data.
+- The complete Jepsen unit suite passed: 207 tests, 579 assertions, zero failures/errors.
+- Full local `go test ./...` is not a pass: the pre-existing FUSE test references Linux-only `syscall.Statfs_t.Namelen` on macOS. Linux CI, the new encryption on/off Jepsen runs, and review of the acceptance changes remain gates.
+
+The design remains Partial until those gates and the process-restart evidence are reconciled. Adding tests or changing the status alone does not complete acceptance.
diff --git a/jepsen/src/elastickv/dynamodb_seal_workload.clj b/jepsen/src/elastickv/dynamodb_seal_workload.clj
new file mode 100644
index 000000000..96140754e
--- /dev/null
+++ b/jepsen/src/elastickv/dynamodb_seal_workload.clj
@@ -0,0 +1,90 @@
+(ns elastickv.dynamodb-seal-workload
+  "Local DynamoDB append acceptance with recorded, verified Raft elections."
+  (:gen-class)
+  (:require [clojure.java.shell :as shell]
+            [clojure.string :as str]
+            [elastickv.cli :as cli]
+            [elastickv.dynamodb-workload :as dynamo]
+            [jepsen.checker :as checker]
+            [jepsen.generator :as gen]
+            [jepsen.nemesis :as nemesis]))
+
+(defn- admin! [binary address command]
+  (let [result (shell/sh binary address command)]
+    (when-not (zero? (:exit result))
+      (throw (ex-info "raftadmin failed" (assoc result :address address :command command))))
+    (:out result)))
+
+(defn- current-leader [binary addresses]
+  (or (some (fn [address]
+              (when (= "state: LEADER" (str/trim (admin! binary address "state")))
+                address))
+            addresses)
+      (throw (ex-info "no Raft leader" {:addresses addresses}))))
+
+(defn leader-transfer-nemesis [binary addresses]
+  (reify
+    nemesis/Reflection
+    (fs [_] #{:leader-transfer})
+    nemesis/Nemesis
+    (setup! [this _] this)
+    (invoke! [_ _ op]
+      (try
+        (let [before (current-leader binary addresses)
+              _ (admin! binary before "leadership_transfer")
+              deadline (+ (System/nanoTime) 10000000000)
+              after (loop []
+                      (let [leader (try (current-leader binary addresses)
+                                        (catch Exception _ nil))]
+                        (cond
+                          (and leader (not= before leader)) leader
+                          (< (System/nanoTime) deadline) (do (Thread/sleep 50) (recur))
+                          :else (throw (ex-info "leadership did not change" {:before before})))))]
+          (assoc op :value {:from before :to after}))
+        (catch Exception e
+          (assoc op :error (.getMessage e) :value :transfer-failed))))
+    (teardown! [_ _] nil)))
+
+(defn leader-change-checker []
+  (reify checker/Checker
+    (check [_ _ history _]
+      (let [ops (filter #(and (= :nemesis (:process %)) (= :leader-transfer (:f %))) history)
+            failures (filter :error ops)
+            changes (filter #(let [{:keys [from to]} (:value %)]
+                               (and from to (not= from to))) ops)]
+        {:valid? (boolean (and (seq changes) (empty? failures)))
+         :verified-leader-changes (count changes)
+         :failures (vec failures)}))))
+
+(defn seal-test [opts]
+  (when-not (and (:local opts) (seq (:raftadmin-bin opts)) (seq (:raft-addresses opts)))
+    (throw (ex-info "seal acceptance requires local mode, raftadmin-bin and raft-addresses" {})))
+  (let [base (dynamo/elastickv-dynamodb-test opts)
+        workload (dynamo/dynamodb-append-workload opts)
+        interval (or (:leader-change-interval opts) 2)]
+    (assoc base
+      :name "elastickv-dynamodb-snapshot-seal"
+      :nemesis (leader-transfer-nemesis (:raftadmin-bin opts) (:raft-addresses opts))
+      :checker (checker/compose {:append (:checker workload) :leader-changes (leader-change-checker)})
+      :generator (->> (:generator workload)
+                      (gen/nemesis (cycle [(gen/sleep interval) {:type :info :f :leader-transfer}]))
+                      (gen/stagger (/ (double (or (:rate opts) 5))))
+                      (gen/time-limit (or (:time-limit opts) 30))))))
+
+(def seal-cli-opts
+  [[nil "--raftadmin-bin PATH" "Repository raftadmin binary (local control only)."]
+   [nil "--raft-addresses ADDRESSES" "Comma-separated local Raft/gRPC endpoints."
+    :parse-fn #(str/split % #",")]
+   [nil "--leader-change-interval SECONDS" "Seconds between leadership transfers."
+    :default 2 :parse-fn #(Double/parseDouble %)
+    :validate [pos? "must be positive"]]])
+
+(defn -main [& args]
+  (cli/run-workload!
+    args (into (into cli/common-cli-opts dynamo/dynamo-cli-opts) seal-cli-opts)
+    (fn [options]
+      (let [options (cli/parse-common-opts options (:dynamo-ports options))]
+        (assoc options
+          :dynamo-host (:host options)
+          :node->port (cli/ports->node-map (:dynamo-ports options) (:nodes options)))))
+    seal-test))
diff --git a/jepsen/test/elastickv/dynamodb_seal_workload_test.clj b/jepsen/test/elastickv/dynamodb_seal_workload_test.clj
new file mode 100644
index 000000000..7706bdea7
--- /dev/null
+++ b/jepsen/test/elastickv/dynamodb_seal_workload_test.clj
@@ -0,0 +1,34 @@
+(ns elastickv.dynamodb-seal-workload-test
+  (:require [clojure.java.shell :as shell]
+            [clojure.test :refer :all]
+            [elastickv.dynamodb-seal-workload :as seal]
+            [jepsen.checker :as checker]
+            [jepsen.nemesis :as nemesis]))
+
+(deftest requires-recorded-successful-leader-change
+  (let [check #(checker/check (seal/leader-change-checker) {} % {})
+        op {:process :nemesis :type :info :f :leader-transfer}]
+    (is (not (:valid? (check []))))
+    (is (not (:valid? (check [(assoc op :value {:from "a" :to "a"})]))))
+    (is (:valid? (check [(assoc op :value {:from "a" :to "b"})])))
+    (is (false? (:valid? (check [(assoc op :value {:from "a" :to "b"})
+                                (assoc op :error "transfer failed")]))))))
+
+(deftest records-actual-transfer-and-failure
+  (let [leader (atom "a")
+        n (seal/leader-transfer-nemesis "/raftadmin" ["a" "b"])
+        op {:type :info :f :leader-transfer}]
+    (with-redefs [shell/sh (fn [binary address command]
+                            (is (= "/raftadmin" binary))
+                            (case command
+                              "state" {:exit 0 :out (str "state: " (if (= address @leader) "LEADER" "FOLLOWER") "\n")}
+                              "leadership_transfer" (do (reset! leader "b") {:exit 0 :out "ok\n"})))]
+      (is (= {:from "a" :to "b"} (:value (nemesis/invoke! n {} op)))))
+    (with-redefs [shell/sh (fn [& _] {:exit 1 :err "RPC failed"})]
+      (is (= "raftadmin failed" (:error (nemesis/invoke! n {} op)))))))
+
+(deftest acceptance-is-local-and-opt-in
+  (is (thrown? clojure.lang.ExceptionInfo (seal/seal-test {})))
+  (let [test (seal/seal-test {:local true :raftadmin-bin "/raftadmin" :raft-addresses ["a" "b"]})]
+    (is (= "elastickv-dynamodb-snapshot-seal" (:name test)))
+    (is (= #{:leader-transfer} (nemesis/fs (:nemesis test))))))
diff --git a/main.go b/main.go
index 6efe3f584..be7f1abae 100644
--- a/main.go
+++ b/main.go
@@ -1321,6 +1321,11 @@ func validateSQSFifoPartitionMapNoDedicatedTSOGroup(partitionMap map[string]sqsF
 }
 
 func buildLeaderDynamo(groups []groupSpec, dynamoAddr string, raftDynamoMap string) (map[string]string, error) {
+	// An absent HTTP map selects the coordinator's gRPC follower path. A
+	// self-only map cannot resolve a remote leader and rejects every request.
+	if strings.TrimSpace(raftDynamoMap) == "" {
+		return nil, nil
+	}
 	return buildLeaderAddrMap(groups, dynamoAddr, raftDynamoMap, parseRaftDynamoMap)
 }
 
diff --git a/main_dynamodb_follower_test.go b/main_dynamodb_follower_test.go
new file mode 100644
index 000000000..b983093f4
--- /dev/null
+++ b/main_dynamodb_follower_test.go
@@ -0,0 +1,25 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestBuildLeaderDynamoWithoutHTTPMap(t *testing.T) {
+	t.Parallel()
+	groups := []groupSpec{{id: 1, address: "127.0.0.1:50051"}}
+	for _, raw := range []string{"", "  "} {
+		got, err := buildLeaderDynamo(groups, "127.0.0.1:8000", raw)
+		require.NoError(t, err)
+		require.Empty(t, got)
+	}
+	got, err := buildLeaderDynamo(groups, "127.0.0.1:8000", "127.0.0.1:50052=127.0.0.1:8001")
+	require.NoError(t, err)
+	require.Equal(t, map[string]string{
+		"127.0.0.1:50051": "127.0.0.1:8000",
+		"127.0.0.1:50052": "127.0.0.1:8001",
+	}, got)
+	_, err = buildLeaderDynamo(groups, "127.0.0.1:8000", "invalid")
+	require.Error(t, err)
+}

From 2205e678f71215312f851d98b9e2b67f8688456d Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Thu, 22 Oct 2026 23:34:23 +0900
Subject: [PATCH 2/5] fix: isolate seal histories and retry forwarded locks

---
 adapter/dynamodb_snapshot_seal_raft_test.go   |  8 +--
 adapter/dynamodb_write_error.go               | 15 +++--
 adapter/dynamodb_write_error_test.go          | 67 ++++++++++++++++---
 ...26_10_11_partial_dynamodb_snapshot_seal.md | 21 ++++--
 .../src/elastickv/dynamodb_seal_workload.clj  | 20 ++++++
 .../elastickv/dynamodb_seal_workload_test.clj | 32 +++++++++
 6 files changed, 139 insertions(+), 24 deletions(-)

diff --git a/adapter/dynamodb_snapshot_seal_raft_test.go b/adapter/dynamodb_snapshot_seal_raft_test.go
index 814c48dea..4d421acf4 100644
--- a/adapter/dynamodb_snapshot_seal_raft_test.go
+++ b/adapter/dynamodb_snapshot_seal_raft_test.go
@@ -83,8 +83,8 @@ func sealRaftAppend(table, value string) updateItemInput {
 	return in
 }
 
-func sealRaftTxnUpdate(table, value string) transactWriteItem {
-	in := sealRaftAppend(table, value)
+func sealRaftTxnUpdate(table string) transactWriteItem {
+	in := sealRaftAppend(table, "3")
 	return transactWriteItem{Update: &transactUpdateInput{
 		TableName: in.TableName, Key: in.Key, UpdateExpression: in.UpdateExpression,
 		ExpressionAttributeNames: in.ExpressionAttributeNames, ExpressionAttributeValues: in.ExpressionAttributeValues,
@@ -180,7 +180,7 @@ func TestDynamoDBSnapshotSealRealRaftLeaderChangeFollowerIngress(t *testing.T) {
 	txnDone := make(chan error, 1)
 	go func() {
 		txnDone <- sealRaftHTTPCall(ctx, strings.TrimPrefix(httpServer.URL, "http://"), "TransactWriteItems", transactWriteItemsInput{
-			TransactItems: []transactWriteItem{sealRaftTxnUpdate(schema.TableName, "3")},
+			TransactItems: []transactWriteItem{sealRaftTxnUpdate(schema.TableName)},
 		})
 	}()
 	var sealed uint64
@@ -256,7 +256,7 @@ func TestDynamoDBSnapshotSealRealRaftTwoDataGroups(t *testing.T) {
 	second, secondCoord, releaseSecond := newDelayedSealRaftServer(t, server, true)
 	defer releaseSecond()
 	input := transactWriteItemsInput{TransactItems: []transactWriteItem{
-		sealRaftTxnUpdate(left.TableName, "3"), sealRaftTxnUpdate(right.TableName, "3"),
+		sealRaftTxnUpdate(left.TableName), sealRaftTxnUpdate(right.TableName),
 	}}
 	body, err := json.Marshal(input)
 	require.NoError(t, err)
diff --git a/adapter/dynamodb_write_error.go b/adapter/dynamodb_write_error.go
index 51ab9f3c4..d16c31e6c 100644
--- a/adapter/dynamodb_write_error.go
+++ b/adapter/dynamodb_write_error.go
@@ -3,6 +3,7 @@ package adapter
 import (
 	"strings"
 
+	"github.com/bootjp/elastickv/kv"
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
 	"google.golang.org/grpc/codes"
@@ -12,8 +13,8 @@ import (
 // Forwarded OCC conflicts lose their Go error chain. Restore only the known
 // wire envelope so item retries retain the existing exact-timestamp dedup path.
 func normalizeDynamoWriteDispatchError(err error) error {
-	if err == nil || errors.Is(err, store.ErrWriteConflict) {
-		return err
+	if err == nil {
+		return nil
 	}
 	type statusCarrier interface{ GRPCStatus() *status.Status }
 	var carrier statusCarrier
@@ -21,9 +22,13 @@ func normalizeDynamoWriteDispatchError(err error) error {
 		return err
 	}
 	st := carrier.GRPCStatus()
-	if st != nil && (st.Code() == codes.Unknown || st.Code() == codes.Aborted) &&
-		strings.HasPrefix(st.Message(), "key: ") && strings.HasSuffix(st.Message(), ": "+store.ErrWriteConflict.Error()) {
-		return errors.WithStack(errors.Mark(err, store.ErrWriteConflict))
+	if st == nil || (st.Code() != codes.Unknown && st.Code() != codes.Aborted) || !strings.HasPrefix(st.Message(), "key: ") {
+		return err
+	}
+	for _, sentinel := range []error{store.ErrWriteConflict, kv.ErrTxnLocked} {
+		if strings.HasSuffix(st.Message(), ": "+sentinel.Error()) {
+			return errors.WithStack(errors.Mark(err, sentinel))
+		}
 	}
 	return err
 }
diff --git a/adapter/dynamodb_write_error_test.go b/adapter/dynamodb_write_error_test.go
index 5de6e53a8..3ddcb6da4 100644
--- a/adapter/dynamodb_write_error_test.go
+++ b/adapter/dynamodb_write_error_test.go
@@ -1,8 +1,11 @@
 package adapter
 
 import (
+	"context"
+	"sync/atomic"
 	"testing"
 
+	"github.com/bootjp/elastickv/kv"
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
@@ -15,24 +18,30 @@ func TestNormalizeDynamoWriteDispatchError(t *testing.T) {
 	for _, tc := range []struct {
 		name string
 		err  error
-		want bool
+		want error
 	}{
-		{"nil", nil, false},
-		{"local conflict", store.NewWriteConflictError([]byte("k")), true},
-		{"unknown conflict", status.Error(codes.Unknown, "key: k: write conflict"), true},
-		{"wrapped aborted conflict", errors.WithStack(status.Error(codes.Aborted, "key: k: write conflict")), true},
-		{"plain text", errors.New("key: k: write conflict"), false},
-		{"missing key envelope", status.Error(codes.Unknown, "write conflict"), false},
-		{"wrong code", status.Error(codes.Internal, "key: k: write conflict"), false},
-		{"embedded sentinel", status.Error(codes.Unknown, "key: write conflict: invalid value"), false},
+		{"nil", nil, nil},
+		{"local conflict", store.NewWriteConflictError([]byte("k")), store.ErrWriteConflict},
+		{"unknown conflict", status.Error(codes.Unknown, "key: k: write conflict"), store.ErrWriteConflict},
+		{"wrapped aborted conflict", errors.WithStack(status.Error(codes.Aborted, "key: k: write conflict")), store.ErrWriteConflict},
+		{"local lock", kv.NewTxnLockedError([]byte("k")), kv.ErrTxnLocked},
+		{"unknown lock", status.Error(codes.Unknown, "key: k: txn locked"), kv.ErrTxnLocked},
+		{"wrapped aborted lock", errors.WithStack(status.Error(codes.Aborted, "key: k (pending): txn locked")), kv.ErrTxnLocked},
+		{"plain text", errors.New("key: k: write conflict"), nil},
+		{"missing key envelope", status.Error(codes.Unknown, "write conflict"), nil},
+		{"wrong code", status.Error(codes.Internal, "key: k: write conflict"), nil},
+		{"embedded sentinel", status.Error(codes.Unknown, "key: write conflict: invalid value"), nil},
+		{"plain lock", status.Error(codes.Unknown, "txn locked"), nil},
+		{"wrong lock code", status.Error(codes.Internal, "key: k: txn locked"), nil},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			normalized := normalizeDynamoWriteDispatchError(tc.err)
-			require.Equal(t, tc.want, errors.Is(normalized, store.ErrWriteConflict))
 			switch {
-			case tc.want:
+			case tc.want != nil:
+				require.True(t, errors.Is(normalized, tc.want), "missing normalized sentinel %v: %v", tc.want, normalized)
 				require.True(t, shouldPreserveTransactWriteAttempt(normalized))
 			case tc.err != nil:
+				require.False(t, isRetryableTransactWriteError(normalized))
 				require.Same(t, tc.err, normalized)
 			default:
 				require.NoError(t, normalized)
@@ -40,3 +49,39 @@ func TestNormalizeDynamoWriteDispatchError(t *testing.T) {
 		})
 	}
 }
+
+type forwardedDynamoLockCoordinator struct {
+	kv.Coordinator
+	rejected atomic.Bool
+}
+
+func (c *forwardedDynamoLockCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
+	if req != nil && req.IsTxn && !isDynamoSnapshotSealRequest(req) && c.rejected.CompareAndSwap(false, true) {
+		return nil, status.Error(codes.Unknown, "key: pending: txn locked")
+	}
+	return c.Coordinator.Dispatch(ctx, req)
+}
+
+func TestDynamoDBRetriesForwardedTxnLock(t *testing.T) {
+	t.Parallel()
+	for _, txn := range []bool{false, true} {
+		t.Run(map[bool]string{false: "item", true: "transaction"}[txn], func(t *testing.T) {
+			st := store.NewMVCCStore()
+			defer st.Close()
+			schema := dedupItemTable()
+			seedDedupItem(t, st, schema, "1")
+			coord := &forwardedDynamoLockCoordinator{Coordinator: newOCCAdapterCoordinator(st)}
+			server := NewDynamoDBServer(nil, st, coord)
+			if txn {
+				require.NoError(t, server.transactWriteItemsWithRetry(context.Background(), transactWriteItemsInput{
+					TransactItems: []transactWriteItem{sealRaftTxnUpdate(schema.TableName)},
+				}))
+			} else {
+				_, err := server.updateItemWithRetry(context.Background(), appendListInput())
+				require.NoError(t, err)
+			}
+			require.True(t, coord.rejected.Load())
+			require.Equal(t, []string{"1", "3"}, readListValues(t, server, schema))
+		})
+	}
+}
diff --git a/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md b/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
index fcb3078e6..060249e0d 100644
--- a/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
+++ b/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
@@ -102,7 +102,7 @@ The audit above describes main at `429f5ee7`; the following additions address it
 
 These tests exposed two runtime gaps, rather than a missing seal implementation:
 
-1. Forwarded write conflicts arrive as gRPC statuses without the local sentinel chain. [`normalizeDynamoWriteDispatchError`](../../adapter/dynamodb_write_error.go) recognizes only the known key-conflict envelope and restores the sentinel at DynamoDB's item/transaction dispatch boundaries. Existing dedup and recompute behavior remains in place. The shared retry predicates, including their SQS callers, are unchanged.
+1. Forwarded write conflicts and transaction locks arrive as gRPC statuses without the local sentinel chain. [`normalizeDynamoWriteDispatchError`](../../adapter/dynamodb_write_error.go) recognizes only known `Unknown`/`Aborted` key-conflict/key-lock envelopes and restores the sentinel at DynamoDB's item/transaction dispatch boundaries. Existing dedup and recompute behavior remains in place. `TestDynamoDBRetriesForwardedTxnLock` covers item and transaction retries. The shared retry predicates, including their SQS callers, are unchanged.
 2. [`buildLeaderDynamo`](../../main.go) previously synthesized a self-only HTTP map even when the flag was omitted. A follower could not resolve a remote leader in that map. An omitted DynamoDB map now selects the existing coordinator gRPC path; explicit maps retain the previous local-entry completion. [`main_dynamodb_follower_test.go`](../../main_dynamodb_follower_test.go) covers both modes and malformed configuration. Redis, S3, and SQS map builders are unchanged.
 
 All production callers of the changed functions were audited with `rg`: three `commitItemWrite` callers (legacy, first dedup attempt, reused dedup attempt), one `runTransactWriteAttempt` caller, and one `buildLeaderDynamo` caller in runtime configuration.
@@ -110,6 +110,7 @@ All production callers of the changed functions were audited with `rg`: three `c
 ### Follower-ingress Jepsen evidence
 
 [`dynamodb_seal_workload.clj`](../../jepsen/src/elastickv/dynamodb_seal_workload.clj) reuses the existing DynamoDB append workload and Jepsen generator/checker library. Its local nemesis invokes the repository's `raftadmin`, verifies that leadership actually moves to a different node, and records the endpoints in the history. A separate checker rejects zero verified transfers or any failed transfer; a successful append checker alone is insufficient.
+Its DB setup deletes the previous `jepsen_append` table once before any client setup begins; the existing clients then recreate an empty generation. Only a missing table is ignored. This prevents a prior append history on the same cluster from contaminating the new check.
 
 The local run `20261022T231204.299+0900` used three separate server processes on loopback, no `raftDynamoMap`, 15 seconds, rate 30, concurrency 9, maximum transaction length 4, and a two-second transfer interval. The checker output was:
 
@@ -123,6 +124,18 @@ The raw history, `test.jepsen`, checker results, and workload log remain in `jep
 
 The [`Jepsen CI workflow`](../../.github/workflows/jepsen-test.yml) now omits the HTTP leader map and runs this workload in both encryption legs. Its artifact retains the seal history and tested revision even on success. Those new CI executions must be checked separately from the local run.
 
+### OS-process snapshot/restart evidence
+
+At source revision `69f2d932d03dfb45a1cd535418f8bbb076d8a43c`, a second local run (`20261022T232208.506+0900`) enabled `ELASTICKV_RAFT_SNAPSHOT_COUNT=20` on three separate server processes. The append and leader-change checkers passed with six verified transfers and no failed transfers. After the workload stopped, all three processes were terminated and reopened from their original data directories and bootstrap membership. The pre/post statuses were:
+
+| Node | Snapshot index before/after | Applied index before/after | Term before/after |
+| --- | ---: | ---: | ---: |
+| n1 | 401 / 401 | 404 / 407 | 8 / 9 |
+| n2 | 402 / 402 | 404 / 407 | 8 / 9 |
+| n3 | 402 / 402 | 404 / 407 | 8 / 9 |
+
+Consistent scans before and after restart matched all 17 items, including list order and values. The per-node logs, topology, and complete pre/post status plus item JSON are retained under `/private/tmp/elastickv-seal-jepsen-luUUDo/`; the Jepsen history remains in the acceptance worktree. The server binary SHA-256 was `6260694477d19fb7c6c9628accebfa5dcacbe6761ba7848e822b84420ca7f022`. The Go snapshot regression separately checks exact seal-timestamp preservation and stale-write rejection; the process experiment checks the persisted real-cluster data path, not just an in-memory restore.
+
 ### Contention measurements
 
 [`BenchmarkDynamoDBSnapshotSealRealRaft`](../../adapter/dynamodb_snapshot_seal_benchmark_test.go) uses a durable single-voter Raft engine, four independent frontends, and constant-size PutItem replacements. The independent case uses four different buckets; the collision case searches for four different item keys mapping to one actual digest bucket. A hot-item case uses one item across frontends. Local item locks are not shared between frontends.
@@ -141,7 +154,7 @@ All measured operations completed. Collision/hot-item contention increased retri
 
 - The real-Raft tests, all DynamoDB and existing one-phase dedup tests passed with `-race`; the leader-change case also passed ten consecutive runs.
 - `TestBuildLeaderDynamoWithoutHTTPMap` and scoped main/adapter lint passed. Lint used the repository's Go 1.26.6 toolchain because the locally installed analyzer cannot consume Go 1.27 export data.
-- The complete Jepsen unit suite passed: 207 tests, 579 assertions, zero failures/errors.
-- Full local `go test ./...` is not a pass: the pre-existing FUSE test references Linux-only `syscall.Statfs_t.Namelen` on macOS. Linux CI, the new encryption on/off Jepsen runs, and review of the acceptance changes remain gates.
+- The complete Jepsen unit suite after review fixes passed: 209 tests, 586 assertions, zero failures/errors.
+- Full local `go test ./...` is not a pass: the pre-existing FUSE test references Linux-only `syscall.Statfs_t.Namelen` on macOS, an unchanged backup renewal test hit its deadline (five isolated reruns passed), and vet overlapped a transient test edit (subsequent target runs passed). Linux CI, the new encryption on/off Jepsen runs, and current-HEAD review of the acceptance changes remain gates.
 
-The design remains Partial until those gates and the process-restart evidence are reconciled. Adding tests or changing the status alone does not complete acceptance.
+The design remains Partial until those CI and review gates are reconciled. Adding tests or changing the status alone does not complete acceptance.
diff --git a/jepsen/src/elastickv/dynamodb_seal_workload.clj b/jepsen/src/elastickv/dynamodb_seal_workload.clj
index 96140754e..e9bcf0987 100644
--- a/jepsen/src/elastickv/dynamodb_seal_workload.clj
+++ b/jepsen/src/elastickv/dynamodb_seal_workload.clj
@@ -3,9 +3,12 @@
   (:gen-class)
   (:require [clojure.java.shell :as shell]
             [clojure.string :as str]
+            [cognitect.aws.client.api :as aws]
             [elastickv.cli :as cli]
             [elastickv.dynamodb-workload :as dynamo]
             [jepsen.checker :as checker]
+            [jepsen.client :as client]
+            [jepsen.db :as db]
             [jepsen.generator :as gen]
             [jepsen.nemesis :as nemesis]))
 
@@ -56,6 +59,22 @@
          :verified-leader-changes (count changes)
          :failures (vec failures)}))))
 
+(defn fresh-append-db []
+  (reify db/DB
+    (setup! [_ test node]
+      ;; DB setup completes on every node before any client setup starts.
+      ;; Delete once; the existing clients recreate an empty new generation.
+      (when (= node (first (:nodes test)))
+        (let [opened (client/open! (:client test) test node)]
+          (try
+            (let [response (aws/invoke (:ddb opened)
+                                      {:op :DeleteTable :request {:TableName "jepsen_append"}})]
+              (when (and (:cognitect.anomalies/category response)
+                         (not= "ResourceNotFoundException" (:__type response)))
+                (throw (ex-info "cannot reset append table" {:response response}))))
+            (finally (client/close! opened test))))))
+    (teardown! [_ _ _] nil)))
+
 (defn seal-test [opts]
   (when-not (and (:local opts) (seq (:raftadmin-bin opts)) (seq (:raft-addresses opts)))
     (throw (ex-info "seal acceptance requires local mode, raftadmin-bin and raft-addresses" {})))
@@ -64,6 +83,7 @@
         interval (or (:leader-change-interval opts) 2)]
     (assoc base
       :name "elastickv-dynamodb-snapshot-seal"
+      :db (fresh-append-db)
       :nemesis (leader-transfer-nemesis (:raftadmin-bin opts) (:raft-addresses opts))
       :checker (checker/compose {:append (:checker workload) :leader-changes (leader-change-checker)})
       :generator (->> (:generator workload)
diff --git a/jepsen/test/elastickv/dynamodb_seal_workload_test.clj b/jepsen/test/elastickv/dynamodb_seal_workload_test.clj
index 7706bdea7..3931b56fc 100644
--- a/jepsen/test/elastickv/dynamodb_seal_workload_test.clj
+++ b/jepsen/test/elastickv/dynamodb_seal_workload_test.clj
@@ -1,8 +1,11 @@
 (ns elastickv.dynamodb-seal-workload-test
   (:require [clojure.java.shell :as shell]
             [clojure.test :refer :all]
+            [cognitect.aws.client.api :as aws]
             [elastickv.dynamodb-seal-workload :as seal]
             [jepsen.checker :as checker]
+            [jepsen.client :as client]
+            [jepsen.db :as db]
             [jepsen.nemesis :as nemesis]))
 
 (deftest requires-recorded-successful-leader-change
@@ -32,3 +35,32 @@
   (let [test (seal/seal-test {:local true :raftadmin-bin "/raftadmin" :raft-addresses ["a" "b"]})]
     (is (= "elastickv-dynamodb-snapshot-seal" (:name test)))
     (is (= #{:leader-transfer} (nemesis/fs (:nemesis test))))))
+
+(deftest resets-the-append-table-once-before-client-setup
+  (let [calls (atom [])
+        closed (atom 0)
+        test {:nodes ["n1" "n2"] :client ::client}]
+    (with-redefs [client/open! (fn [& _] {:ddb ::ddb})
+                  client/close! (fn [& _] (swap! closed inc))
+                  aws/invoke (fn [ddb request]
+                               (is (= ::ddb ddb))
+                               (swap! calls conj request)
+                               {})]
+      (doseq [node (:nodes test)] (db/setup! (seal/fresh-append-db) test node))
+      (is (= [{:op :DeleteTable :request {:TableName "jepsen_append"}}] @calls))
+      (is (= 1 @closed)))))
+
+(deftest reset-only-ignores-missing-table-and-always-closes
+  (doseq [[response fails?] [[{:cognitect.anomalies/category :not-found
+                             :__type "ResourceNotFoundException"} false]
+                           [{:cognitect.anomalies/category :fault
+                             :__type "InternalServerError"} true]]]
+    (let [closed (atom false)]
+      (with-redefs [client/open! (fn [& _] {:ddb ::ddb})
+                    client/close! (fn [& _] (reset! closed true))
+                    aws/invoke (fn [& _] response)]
+        (let [run #(db/setup! (seal/fresh-append-db) {:nodes ["n1"]} "n1")]
+          (if fails?
+            (is (thrown? clojure.lang.ExceptionInfo (run)))
+            (is (nil? (run)))))
+        (is @closed)))))

From 8db0155d632aec73a433cc3478f0a474c6921e8d Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Thu, 22 Oct 2026 23:45:31 +0900
Subject: [PATCH 3/5] fix: normalize all DynamoDB forwarded write failures

---
 adapter/dynamodb_item_write.go                |  4 +-
 adapter/dynamodb_locks.go                     |  3 +-
 adapter/dynamodb_migration.go                 |  6 +--
 adapter/dynamodb_schema.go                    |  7 +--
 adapter/dynamodb_snapshot_seal_raft_test.go   | 29 ++++++++++++
 adapter/dynamodb_transact.go                  |  3 +-
 adapter/dynamodb_write_error.go               |  6 +++
 adapter/dynamodb_write_error_test.go          | 46 ++++++++++++++++++-
 ...26_10_11_partial_dynamodb_snapshot_seal.md |  3 +-
 9 files changed, 94 insertions(+), 13 deletions(-)

diff --git a/adapter/dynamodb_item_write.go b/adapter/dynamodb_item_write.go
index 1354ff184..3406d1743 100644
--- a/adapter/dynamodb_item_write.go
+++ b/adapter/dynamodb_item_write.go
@@ -483,9 +483,9 @@ func (d *DynamoDBServer) preparePutItemWrite(ctx context.Context, in putItemInpu
 
 func (d *DynamoDBServer) commitItemWrite(ctx context.Context, readTimestamp kv.ReadTimestamp, req *kv.OperationGroup[kv.OP]) error {
 	dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
-	_, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req)
+	_, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req)
 	if err != nil {
-		return errors.WithStack(normalizeDynamoWriteDispatchError(err))
+		return errors.WithStack(err)
 	}
 	return nil
 }
diff --git a/adapter/dynamodb_locks.go b/adapter/dynamodb_locks.go
index df2375c79..75b1bcd87 100644
--- a/adapter/dynamodb_locks.go
+++ b/adapter/dynamodb_locks.go
@@ -391,8 +391,9 @@ func (d *DynamoDBServer) tryTxnWriteTimestamp(ctx context.Context, label string,
 		var resp *kv.CoordinateResponse
 		if sealTS == 0 {
 			resp, err = kv.DispatchWithLeaderIssuedCommitTS(floor.WithDispatchVoucher(ctx), d.coordinator, request)
+			err = normalizeDynamoWriteDispatchError(err)
 		} else {
-			resp, err = kv.DispatchWithReadTimestamp(floor.WithDispatchVoucher(ctx), d.coordinator, request)
+			resp, err = d.dispatchWriteWithReadTimestamp(floor.WithDispatchVoucher(ctx), request)
 		}
 		if err != nil {
 			return kv.ReadTimestamp{}, errors.WithStack(err)
diff --git a/adapter/dynamodb_migration.go b/adapter/dynamodb_migration.go
index e7b51b1f7..d3e9f408a 100644
--- a/adapter/dynamodb_migration.go
+++ b/adapter/dynamodb_migration.go
@@ -97,7 +97,7 @@ func (d *DynamoDBServer) startLegacyTableKeyMigration(
 	}
 	req.StartTS = readTS
 	dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
-	if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err != nil {
+	if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err != nil {
 		return errors.WithStack(err)
 	}
 	return nil
@@ -186,7 +186,7 @@ func (d *DynamoDBServer) migrateLegacyItem(
 		req.StartTS = readTS
 		req.ReadKeys = append(req.ReadKeys, sealKey)
 		dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
-		if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err == nil {
+		if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err == nil {
 			return nil
 		} else if !isRetryableTransactWriteError(err) {
 			return errors.WithStack(err)
@@ -331,7 +331,7 @@ func (d *DynamoDBServer) finalizeLegacyTableMigration(ctx context.Context, schem
 		},
 	}
 	dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
-	if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err != nil {
+	if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err != nil {
 		return errors.WithStack(err)
 	}
 	d.launchDeletedTableCleanup(schema.TableName, oldGeneration)
diff --git a/adapter/dynamodb_schema.go b/adapter/dynamodb_schema.go
index d3565477a..fedba2556 100644
--- a/adapter/dynamodb_schema.go
+++ b/adapter/dynamodb_schema.go
@@ -205,7 +205,7 @@ func (d *DynamoDBServer) createTableWithRetry(ctx context.Context, tableName str
 		}
 		req.StartTS = readTS
 		dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
-		if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err == nil {
+		if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err == nil {
 			return nil
 		} else if !isRetryableTransactWriteError(err) {
 			return errors.WithStack(err)
@@ -319,7 +319,7 @@ func (d *DynamoDBServer) deleteTableWithRetry(ctx context.Context, tableName str
 			},
 		}
 		dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
-		if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err != nil {
+		if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, req); err != nil {
 			if !isRetryableTransactWriteError(err) {
 				return errors.WithStack(err)
 			}
@@ -377,6 +377,7 @@ func (d *DynamoDBServer) dispatchDeletedTableCleanupPrefix(ctx context.Context,
 				{Op: kv.DelPrefix, Key: prefix},
 			},
 		})
+		err = normalizeDynamoWriteDispatchError(err)
 		if err == nil {
 			return nil
 		}
@@ -403,7 +404,7 @@ func (d *DynamoDBServer) dispatchDeleteBatch(ctx context.Context, keys [][]byte)
 	}
 	_, err := d.coordinator.Dispatch(ctx, req)
 	if err != nil {
-		return errors.WithStack(err)
+		return errors.WithStack(normalizeDynamoWriteDispatchError(err))
 	}
 	return nil
 }
diff --git a/adapter/dynamodb_snapshot_seal_raft_test.go b/adapter/dynamodb_snapshot_seal_raft_test.go
index 4d421acf4..4833eeb73 100644
--- a/adapter/dynamodb_snapshot_seal_raft_test.go
+++ b/adapter/dynamodb_snapshot_seal_raft_test.go
@@ -216,6 +216,35 @@ func sealRaftAdvanceWatermark(t *testing.T, ctx context.Context, coord kv.Coordi
 	require.NoError(t, err)
 }
 
+func TestDynamoDBRealRaftFollowerConcurrentCreateTable(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	nodes, _, _ := createNode(t, 3)
+	defer shutdown(nodes)
+	input := map[string]any{
+		"TableName":             "concurrent_seal_setup",
+		"AttributeDefinitions":  []map[string]string{{"AttributeName": "pk", "AttributeType": "S"}},
+		"KeySchema":             []map[string]string{{"AttributeName": "pk", "KeyType": "HASH"}},
+		"ProvisionedThroughput": map[string]int{"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+	}
+	const clients = 9
+	done := make(chan error, clients)
+	for i := range clients {
+		address := nodes[1+i%2].dynamoAddress
+		go func() { done <- sealRaftHTTPCall(ctx, address, "CreateTable", input) }()
+	}
+	created := 0
+	for range clients {
+		err := sealRaftReceive(t, ctx, done)
+		if err == nil {
+			created++
+		} else {
+			require.ErrorContains(t, err, "ResourceInUseException", "concurrent setup must not leak a forwarded OCC conflict")
+		}
+	}
+	require.Equal(t, 1, created)
+}
+
 func TestDynamoDBSnapshotSealRealRaftTwoDataGroups(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
 	defer cancel()
diff --git a/adapter/dynamodb_transact.go b/adapter/dynamodb_transact.go
index 217437b2d..9d8553d9f 100644
--- a/adapter/dynamodb_transact.go
+++ b/adapter/dynamodb_transact.go
@@ -752,8 +752,7 @@ func (d *DynamoDBServer) runTransactWriteAttempt(
 		return true, nil
 	}
 	dispatchCtx := reqs.readTimestamp.WithDispatchVoucher(ctx)
-	if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, reqs.group); err != nil {
-		err = normalizeDynamoWriteDispatchError(err)
+	if _, err := d.dispatchWriteWithReadTimestamp(dispatchCtx, reqs.group); err != nil {
 		wrapped := errors.WithStack(err)
 		if !isRetryableTransactWriteError(err) {
 			return false, wrapped
diff --git a/adapter/dynamodb_write_error.go b/adapter/dynamodb_write_error.go
index d16c31e6c..4caaf4554 100644
--- a/adapter/dynamodb_write_error.go
+++ b/adapter/dynamodb_write_error.go
@@ -1,6 +1,7 @@
 package adapter
 
 import (
+	"context"
 	"strings"
 
 	"github.com/bootjp/elastickv/kv"
@@ -10,6 +11,11 @@ import (
 	"google.golang.org/grpc/status"
 )
 
+func (d *DynamoDBServer) dispatchWriteWithReadTimestamp(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
+	resp, err := kv.DispatchWithReadTimestamp(ctx, d.coordinator, req)
+	return resp, normalizeDynamoWriteDispatchError(err)
+}
+
 // Forwarded OCC conflicts lose their Go error chain. Restore only the known
 // wire envelope so item retries retain the existing exact-timestamp dedup path.
 func normalizeDynamoWriteDispatchError(err error) error {
diff --git a/adapter/dynamodb_write_error_test.go b/adapter/dynamodb_write_error_test.go
index 3ddcb6da4..3612f6750 100644
--- a/adapter/dynamodb_write_error_test.go
+++ b/adapter/dynamodb_write_error_test.go
@@ -53,15 +53,59 @@ func TestNormalizeDynamoWriteDispatchError(t *testing.T) {
 type forwardedDynamoLockCoordinator struct {
 	kv.Coordinator
 	rejected atomic.Bool
+	failure  error
+	seal     bool
 }
 
 func (c *forwardedDynamoLockCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
-	if req != nil && req.IsTxn && !isDynamoSnapshotSealRequest(req) && c.rejected.CompareAndSwap(false, true) {
+	if req != nil && req.IsTxn && isDynamoSnapshotSealRequest(req) == c.seal && c.rejected.CompareAndSwap(false, true) {
+		if c.failure != nil {
+			return nil, c.failure
+		}
 		return nil, status.Error(codes.Unknown, "key: pending: txn locked")
 	}
 	return c.Coordinator.Dispatch(ctx, req)
 }
 
+func TestDynamoDBRetriesForwardedSnapshotSealConflict(t *testing.T) {
+	t.Parallel()
+	for _, sentinel := range []error{store.ErrWriteConflict, kv.ErrTxnLocked} {
+		t.Run(sentinel.Error(), func(t *testing.T) {
+			st := store.NewMVCCStore()
+			defer st.Close()
+			schema := dedupItemTable()
+			seedDedupItem(t, st, schema, "1")
+			coord := &forwardedDynamoLockCoordinator{
+				Coordinator: newOCCAdapterCoordinator(st),
+				seal:        true,
+				failure:     status.Error(codes.Aborted, "key: !ddb|seal|pending: "+sentinel.Error()),
+			}
+			server := NewDynamoDBServer(nil, st, coord)
+			_, err := server.updateItemWithRetry(context.Background(), appendListInput())
+			require.NoError(t, err)
+			require.True(t, coord.rejected.Load())
+			require.Equal(t, []string{"1", "3"}, readListValues(t, server, schema))
+		})
+	}
+}
+
+func TestDynamoDBCreateTableRetriesForwardedConflict(t *testing.T) {
+	t.Parallel()
+	st := store.NewMVCCStore()
+	defer st.Close()
+	coord := &forwardedDynamoLockCoordinator{
+		Coordinator: newOCCAdapterCoordinator(st),
+		failure:     status.Error(codes.Unknown, "key: !ddb|meta|table|test: write conflict"),
+	}
+	server := NewDynamoDBServer(nil, st, coord)
+	schema := dedupItemTable()
+	require.NoError(t, server.createTableWithRetry(context.Background(), schema.TableName, schema))
+	require.True(t, coord.rejected.Load())
+	exists, err := server.tableExistsAt(context.Background(), schema.TableName, consistentReadLatestTS)
+	require.NoError(t, err)
+	require.True(t, exists)
+}
+
 func TestDynamoDBRetriesForwardedTxnLock(t *testing.T) {
 	t.Parallel()
 	for _, txn := range []bool{false, true} {
diff --git a/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md b/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
index 060249e0d..33e0c8e98 100644
--- a/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
+++ b/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
@@ -102,10 +102,11 @@ The audit above describes main at `429f5ee7`; the following additions address it
 
 These tests exposed two runtime gaps, rather than a missing seal implementation:
 
-1. Forwarded write conflicts and transaction locks arrive as gRPC statuses without the local sentinel chain. [`normalizeDynamoWriteDispatchError`](../../adapter/dynamodb_write_error.go) recognizes only known `Unknown`/`Aborted` key-conflict/key-lock envelopes and restores the sentinel at DynamoDB's item/transaction dispatch boundaries. Existing dedup and recompute behavior remains in place. `TestDynamoDBRetriesForwardedTxnLock` covers item and transaction retries. The shared retry predicates, including their SQS callers, are unchanged.
+1. Forwarded write conflicts and transaction locks arrive as gRPC statuses without the local sentinel chain. [`normalizeDynamoWriteDispatchError`](../../adapter/dynamodb_write_error.go) recognizes only known `Unknown`/`Aborted` key-conflict/key-lock envelopes. All DynamoDB write dispatches normalize failures: item/transaction commits, snapshot seals, table schema/migration, and cleanup. The common read-timestamp dispatch helper preserves the existing voucher handling; leader-issued seal and nontransactional cleanup dispatches retain their original timestamp paths. Existing dedup and recompute behavior remains in place. `TestDynamoDBRetriesForwardedTxnLock`, `TestDynamoDBRetriesForwardedSnapshotSealConflict`, and `TestDynamoDBCreateTableRetriesForwardedConflict` cover these retry boundaries. The shared retry predicates, including their SQS callers, are unchanged.
 2. [`buildLeaderDynamo`](../../main.go) previously synthesized a self-only HTTP map even when the flag was omitted. A follower could not resolve a remote leader in that map. An omitted DynamoDB map now selects the existing coordinator gRPC path; explicit maps retain the previous local-entry completion. [`main_dynamodb_follower_test.go`](../../main_dynamodb_follower_test.go) covers both modes and malformed configuration. Redis, S3, and SQS map builders are unchanged.
 
 All production callers of the changed functions were audited with `rg`: three `commitItemWrite` callers (legacy, first dedup attempt, reused dedup attempt), one `runTransactWriteAttempt` caller, and one `buildLeaderDynamo` caller in runtime configuration.
+The follow-up dispatch audit covers all eight reusable read-timestamp write sites plus the leader-issued seal and two direct cleanup dispatches; no DynamoDB write dispatch remains outside normalization. `TestDynamoDBRealRaftFollowerConcurrentCreateTable` sends nine concurrent HTTP setup requests through followers and requires one creation plus ResourceInUse responses, never leaked OCC errors; five race-enabled repetitions passed.
 
 ### Follower-ingress Jepsen evidence
 

From f432933c8c166901e9636750e3e879b56cc789b3 Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Fri, 23 Oct 2026 00:03:23 +0900
Subject: [PATCH 4/5] docs: retain encrypted seal acceptance failures

---
 .../2026_10_11_partial_dynamodb_snapshot_seal.md     | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

diff --git a/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md b/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
index 33e0c8e98..2a2731775 100644
--- a/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
+++ b/docs/design/2026_10_11_partial_dynamodb_snapshot_seal.md
@@ -158,4 +158,14 @@ All measured operations completed. Collision/hot-item contention increased retri
 - The complete Jepsen unit suite after review fixes passed: 209 tests, 586 assertions, zero failures/errors.
 - Full local `go test ./...` is not a pass: the pre-existing FUSE test references Linux-only `syscall.Statfs_t.Namelen` on macOS, an unchanged backup renewal test hit its deadline (five isolated reruns passed), and vet overlapped a transient test edit (subsequent target runs passed). Linux CI, the new encryption on/off Jepsen runs, and current-HEAD review of the acceptance changes remain gates.
 
-The design remains Partial until those CI and review gates are reconciled. Adding tests or changing the status alone does not complete acceptance.
+At `8db0155d632aec73a433cc3478f0a474c6921e8d`, current-head code/security reviews had no findings and all required CI passed, including the complete Ubuntu Go suite and encryption off/on Jepsen. The non-root local Linux full-suite attempt was not a pass: an unchanged SQS FIFO visibility test failed under load, then passed twenty isolated repetitions. The separate Ubuntu CI result does not erase that local failure.
+
+### Outstanding encrypted stress findings
+
+The design remains Partial despite the short successful CI and review results:
+
+- Additional encrypted three-process stress at `8db0155d`, with snapshot interval 20, a preceding append workload, fresh-table reset, and six verified leader changes, returned `valid? false` with `G-single-item-realtime` and `G1c-realtime`. An acknowledged append 37 to key 16 was absent from a later successful read including 46. History `20261022T235149.907+0900` and the per-node logs under `/private/tmp/elastickv-seal-jepsen-FpXSbX/` are retained. This run is not counted as a success.
+- Existing [PR #1286](https://github.com/bootjp/elastickv/pull/1286) addresses an overlapping Raft apply/timestamp problem. A diagnostic build combining its `78cab1ba` with `8db0155d` passed one identical encrypted workload with six verified transfers. This suggests overlap, not a causal proof or a completed stress campaign. The comparison was made in a separate worktree; this acceptance PR does not duplicate that implementation.
+- The combined run's subsequent encrypted restart failed: the startup guard attempted to scan Raft entries `(6, 877]` after compaction and MemoryStorage returned `requested index is unavailable due to compaction`. The unmodified `8db0155d` binary also refuses startup on that retained diagnostic state, without the #1286 patch. Logs are under `/private/tmp/elastickv-seal-jepsen-dscUKN/`. The earlier plaintext restart success is not evidence that encrypted restart is safe.
+
+Before declaring this design complete, coordinate the existing Raft fix without duplication, cover and resolve the encrypted compacted-snapshot startup failure, and rerun the acceptance matrix. Adding tests, receiving a clean review, or changing status alone does not complete acceptance.

From a0f55065ac07b53413559e0249cba4697c2feade Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Fri, 23 Oct 2026 01:20:31 +0900
Subject: [PATCH 5/5] test: distinguish snapshot seals from global timestamp
 fences

---
 adapter/dynamodb_snapshot_seal_raft_test.go | 41 ++++++++++++++++++---
 1 file changed, 35 insertions(+), 6 deletions(-)

diff --git a/adapter/dynamodb_snapshot_seal_raft_test.go b/adapter/dynamodb_snapshot_seal_raft_test.go
index 4833eeb73..7a79bfcfd 100644
--- a/adapter/dynamodb_snapshot_seal_raft_test.go
+++ b/adapter/dynamodb_snapshot_seal_raft_test.go
@@ -26,11 +26,12 @@ import (
 // it. No store/FSM/transport is mocked: the later seal must reject this plan.
 type delayedSealRaftCoordinator struct {
 	kv.Coordinator
-	blocked   atomic.Bool
-	entered   chan uint64
-	release   chan struct{}
-	result    chan error
-	unstamped bool
+	blocked       atomic.Bool
+	entered       chan uint64
+	release       chan struct{}
+	result        chan error
+	unstamped     bool
+	freshCommitTS uint64
 }
 
 func (c *delayedSealRaftCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
@@ -47,6 +48,13 @@ func (c *delayedSealRaftCoordinator) Dispatch(ctx context.Context, req *kv.Opera
 	case <-ctx.Done():
 		return nil, ctx.Err()
 	}
+	return c.dispatchReleased(ctx, req)
+}
+
+func (c *delayedSealRaftCoordinator) dispatchReleased(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
+	if c.freshCommitTS != 0 {
+		req.CommitTS = c.freshCommitTS
+	}
 	resp, err := c.Coordinator.Dispatch(ctx, req)
 	c.result <- err
 	return resp, err
@@ -145,6 +153,15 @@ func sealRaftHTTPCall(ctx context.Context, address, action string, input any) er
 }
 
 func TestDynamoDBSnapshotSealRealRaftLeaderChangeFollowerIngress(t *testing.T) {
+	for _, restamp := range []bool{false, true} {
+		t.Run(map[bool]string{false: "delayed-commit-watermark", true: "fresh-commit-stale-read-seal"}[restamp], func(t *testing.T) {
+			testSnapshotSealRaftLeaderChange(t, restamp)
+		})
+	}
+}
+
+func testSnapshotSealRaftLeaderChange(t *testing.T, restamp bool) {
+	t.Helper()
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 	nodes, _, _ := createNode(t, 3)
@@ -192,8 +209,20 @@ func TestDynamoDBSnapshotSealRealRaftLeaderChangeFollowerIngress(t *testing.T) {
 		t.Fatal(ctx.Err())
 	}
 	require.Greater(t, sealed, commitTS)
+	if restamp {
+		// Keep the stale read plan but allocate its commit timestamp after R.
+		// The global apply-order fence must not hide whether the seal rejects it.
+		fresh, err := kv.NextTimestampAfterThrough(ctx, nodes[1].dynamoServer.coordinator, sealed, "seal regression: fresh commit for stale read plan")
+		require.NoError(t, err)
+		require.Greater(t, fresh, sealed)
+		coord.freshCommitTS = fresh
+	}
 	release()
-	require.ErrorContains(t, sealRaftReceive(t, ctx, coord.result), dynamoSealPrefix, "only the seal changed since the first read")
+	conflict := sealRaftReceive(t, ctx, coord.result)
+	require.ErrorContains(t, conflict, "write conflict", "the delayed write must never be acknowledged")
+	if restamp {
+		require.ErrorContains(t, conflict, dynamoSealPrefix, "a fresh commit timestamp bypasses the global fence, so the seal itself must reject the stale read")
+	}
 	require.NoError(t, sealRaftReceive(t, ctx, done))
 	releaseSecond()
 	require.NoError(t, sealRaftReceive(t, ctx, txnDone))