tso: add dedicated ceiling fsm

adapter/dynamodb_item_write.go

@@ -155,7 +155,11 @@ func (d *DynamoDBServer) retryItemWriteWithGenerationLegacy(
 	backoff := transactRetryInitialBackoff
 	deadline := time.Now().Add(transactRetryMaxDuration)
 	for range transactRetryMaxAttempts {
-		readTS := d.nextTxnReadTS()
+		readTimestamp, err := d.beginTxnReadTimestamp(ctx, "dynamodb item-write legacy: begin read timestamp")
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+		readTS := readTimestamp.Timestamp()
 		plan, err := prepare(readTS)
 		if err != nil {
 			return nil, err
@@ -164,7 +168,7 @@ func (d *DynamoDBServer) retryItemWriteWithGenerationLegacy(
 			return plan, nil
 		}
 		plan.req.StartTS = readTS
-		if err = d.commitItemWrite(ctx, plan.req); err != nil {
+		if err = d.commitItemWrite(ctx, readTimestamp, plan.req); err != nil {
 			if !isRetryableTransactWriteError(err) {
 				return nil, errors.WithStack(err)
 			}
@@ -204,6 +208,10 @@ type reusableItemWrite struct {
 	// — the write set was built once from attempt 1's read — so plan is also the
 	// correct value to return when the FSM dedup no-ops the apply (R1).
 	plan *itemWritePlan
+	// readTimestamp carries the Phase-D dispatch capability that validates the
+	// reused StartTS. Retaining only the numeric StartTS would leave retries
+	// unable to re-vouch the same applied read watermark.
+	readTimestamp kv.ReadTimestamp
 	// commitTS is the most recent dispatched commit_ts for this write set; the
 	// next retry passes it as PrevCommitTS so the FSM probes exactly the attempt
 	// that might have landed.
@@ -266,7 +274,11 @@ func (d *DynamoDBServer) itemWriteFirstAttempt(
 	tableName string,
 	prepare func(readTS uint64) (*itemWritePlan, error),
 ) (*itemWritePlan, *reusableItemWrite, error) {
-	readTS := d.nextTxnReadTS()
+	readTimestamp, err := d.beginTxnReadTimestamp(ctx, "dynamodb item-write: begin read timestamp")
+	if err != nil {
+		return nil, nil, errors.WithStack(err)
+	}
+	readTS := readTimestamp.Timestamp()
 	plan, err := prepare(readTS)
 	if err != nil {
 		return nil, nil, err
@@ -283,13 +295,14 @@ func (d *DynamoDBServer) itemWriteFirstAttempt(
 	}
 	plan.req.StartTS = readTS
 	plan.req.CommitTS = commitTS
-	if dispErr := d.commitItemWrite(ctx, plan.req); dispErr != nil {
+	if dispErr := d.commitItemWrite(ctx, readTimestamp, plan.req); dispErr != nil {
 		// dispErr is already wrapped by commitItemWrite; return it raw.
 		if isRetryableTransactWriteError(dispErr) {
 			return nil, &reusableItemWrite{
-				plan:     plan,
-				commitTS: commitTS,
-				probeKey: kv.PrimaryKeyForElems(plan.req.Elems),
+				plan:          plan,
+				readTimestamp: readTimestamp,
+				commitTS:      commitTS,
+				probeKey:      kv.PrimaryKeyForElems(plan.req.Elems),
 			}, dispErr
 		}
 		return nil, nil, dispErr
@@ -311,7 +324,7 @@ func (d *DynamoDBServer) itemWriteReuseAttempt(
 	}
 	pending.plan.req.CommitTS = commitTS
 	pending.plan.req.PrevCommitTS = pending.commitTS
-	dispErr := d.commitItemWrite(ctx, pending.plan.req)
+	dispErr := d.commitItemWrite(ctx, pending.readTimestamp, pending.plan.req)
 	if dispErr == nil {
 		return d.finishItemWriteAttempt(ctx, tableName, pending.plan)
 	}
@@ -433,8 +446,9 @@ func (d *DynamoDBServer) preparePutItemWrite(ctx context.Context, in putItemInpu
 	}, nil
 }
 
-func (d *DynamoDBServer) commitItemWrite(ctx context.Context, req *kv.OperationGroup[kv.OP]) error {
-	_, err := d.coordinator.Dispatch(ctx, req)
+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)
 	if err != nil {
 		return errors.WithStack(err)
 	}

adapter/dynamodb_locks.go

@@ -115,6 +115,11 @@ func (d *DynamoDBServer) nextTxnReadTS() uint64 {
 	return maxTS
 }
 
+func (d *DynamoDBServer) beginTxnReadTimestamp(ctx context.Context, label string) (kv.ReadTimestamp, error) {
+	readTimestamp, err := kv.BeginReadTimestampThrough(ctx, d.coordinator, d.nextTxnReadTS(), label)
+	return readTimestamp, errors.WithStack(err)
+}
+
 func (d *DynamoDBServer) pinReadTS(ts uint64) *kv.ActiveTimestampToken {
 	if d == nil || d.readTracker == nil {
 		return &kv.ActiveTimestampToken{}

adapter/dynamodb_migration.go

@@ -21,12 +21,11 @@ func (d *DynamoDBServer) ensureLegacyTableMigrationLocked(ctx context.Context, t
 	backoff := transactRetryInitialBackoff
 	deadline := time.Now().Add(transactRetryMaxDuration)
 	for range transactRetryMaxAttempts {
-		readTS := d.nextTxnReadTS()
-		schema, exists, err := d.loadTableSchemaAt(ctx, tableName, readTS)
+		readTimestamp, schema, migrationRequired, err := d.legacyMigrationSnapshot(ctx, tableName)
 		if err != nil {
 			return errors.WithStack(err)
 		}
-		if !exists || !schema.needsLegacyKeyMigration() {
+		if !migrationRequired {
 			return nil
 		}
 		// Admin read-only callers (AdminScanTable) must not trigger
@@ -39,7 +38,7 @@ func (d *DynamoDBServer) ensureLegacyTableMigrationLocked(ctx context.Context, t
 				"table requires a one-time legacy-key migration before admin read endpoints are available; migrate via the SigV4 surface first")
 		}
 		if !schema.usesOrderedKeyEncoding() {
-			err = d.startLegacyTableKeyMigration(ctx, schema, readTS)
+			err = d.startLegacyTableKeyMigration(ctx, schema, readTimestamp)
 		} else {
 			err = d.migrateLegacyTableGeneration(ctx, schema)
 		}
@@ -57,14 +56,30 @@ func (d *DynamoDBServer) ensureLegacyTableMigrationLocked(ctx context.Context, t
 	return newDynamoAPIError(http.StatusInternalServerError, dynamoErrInternal, "legacy table migration retry attempts exhausted")
 }
 
+func (d *DynamoDBServer) legacyMigrationSnapshot(
+	ctx context.Context,
+	tableName string,
+) (kv.ReadTimestamp, *dynamoTableSchema, bool, error) {
+	readTimestamp, err := d.beginTxnReadTimestamp(ctx, "dynamodb legacy-table migration: begin read timestamp")
+	if err != nil {
+		return kv.ReadTimestamp{}, nil, false, errors.WithStack(err)
+	}
+	schema, exists, err := d.loadTableSchemaAt(ctx, tableName, readTimestamp.Timestamp())
+	if err != nil {
+		return kv.ReadTimestamp{}, nil, false, errors.WithStack(err)
+	}
+	return readTimestamp, schema, exists && schema.needsLegacyKeyMigration(), nil
+}
+
 func (d *DynamoDBServer) startLegacyTableKeyMigration(
 	ctx context.Context,
 	schema *dynamoTableSchema,
-	readTS uint64,
+	readTimestamp kv.ReadTimestamp,
 ) error {
 	if schema == nil || schema.usesOrderedKeyEncoding() {
 		return nil
 	}
+	readTS := readTimestamp.Timestamp()
 	nextGeneration, err := d.nextTableGenerationAt(ctx, schema.TableName, readTS)
 	if err != nil {
 		return err
@@ -81,7 +96,8 @@ func (d *DynamoDBServer) startLegacyTableKeyMigration(
 		return err
 	}
 	req.StartTS = readTS
-	if _, err := d.coordinator.Dispatch(ctx, req); err != nil {
+	dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
+	if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err != nil {
 		return errors.WithStack(err)
 	}
 	return nil
@@ -151,7 +167,11 @@ func (d *DynamoDBServer) migrateLegacyItem(
 	backoff := transactRetryInitialBackoff
 	deadline := time.Now().Add(transactRetryMaxDuration)
 	for range transactRetryMaxAttempts {
-		readTS := d.nextTxnReadTS()
+		readTimestamp, err := d.beginTxnReadTimestamp(ctx, "dynamodb legacy-item migration: begin read timestamp")
+		if err != nil {
+			return errors.WithStack(err)
+		}
+		readTS := readTimestamp.Timestamp()
 		req, done, err := d.buildLegacyMigrationRequest(ctx, targetSchema, sourceSchema, targetKey, sourceKey, readTS)
 		if err != nil {
 			return err
@@ -160,7 +180,8 @@ func (d *DynamoDBServer) migrateLegacyItem(
 			return nil
 		}
 		req.StartTS = readTS
-		if _, err := d.coordinator.Dispatch(ctx, req); err == nil {
+		dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
+		if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err == nil {
 			return nil
 		} else if !isRetryableTransactWriteError(err) {
 			return errors.WithStack(err)
@@ -292,15 +313,20 @@ func (d *DynamoDBServer) finalizeLegacyTableMigration(ctx context.Context, schem
 	if err != nil {
 		return errors.WithStack(err)
 	}
-	readTS := d.nextTxnReadTS()
+	readTimestamp, err := d.beginTxnReadTimestamp(ctx, "dynamodb finalize migration: begin read timestamp")
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	readTS := readTimestamp.Timestamp()
 	req := &kv.OperationGroup[kv.OP]{
 		IsTxn:   true,
 		StartTS: readTS,
 		Elems: []*kv.Elem[kv.OP]{
 			{Op: kv.Put, Key: dynamoTableMetaKey(schema.TableName), Value: body},
 		},
 	}
-	if _, err := d.coordinator.Dispatch(ctx, req); err != nil {
+	dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
+	if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err != nil {
 		return errors.WithStack(err)
 	}
 	d.launchDeletedTableCleanup(schema.TableName, oldGeneration)

adapter/dynamodb_onephase_dedup_test.go

@@ -133,6 +133,23 @@ func TestItemWriteDedup_PriorAttemptDidNotLand_Applies(t *testing.T) {
 	require.Equal(t, 0, coord.probeNoOps, "nothing landed, so the probe must miss and the reuse applies")
 }
 
+func TestItemWriteDedup_PhaseDVouchesReuse(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	coord := newPhaseDDedupTestCoordinator(st, 1, false)
+	schema, server := newDedupItemWriteServer(st, coord, true)
+	seedDedupItem(t, st, schema, "1", "2")
+
+	plan, err := server.updateItemWithRetry(ctx, appendListInput())
+	require.NoError(t, err)
+	require.NotNil(t, plan)
+
+	require.Equal(t, []string{"1", "2", "3"}, readListValues(t, server, schema))
+	require.Equal(t, 2, coord.dispatches)
+	require.Equal(t, uint64(2), coord.vouches.Load(), "first attempt and reused write set must each reserve a Phase-D dispatch voucher")
+}
+
 // TestItemWriteDedup_SelfInflictedReuseConflict_ReturnsSuccess: attempt 1
 // pre-rejects, the reuse then LANDS but surfaces WriteConflict (self-inflicted
 // conflict under churn). The adapter-side self-conflict guard probes the reuse's

adapter/dynamodb_schema.go

@@ -183,7 +183,11 @@ func (d *DynamoDBServer) createTableWithRetry(ctx context.Context, tableName str
 	backoff := transactRetryInitialBackoff
 	deadline := time.Now().Add(transactRetryMaxDuration)
 	for range transactRetryMaxAttempts {
-		readTS := d.nextTxnReadTS()
+		readTimestamp, err := d.beginTxnReadTimestamp(ctx, "dynamodb create table: begin read timestamp")
+		if err != nil {
+			return errors.WithStack(err)
+		}
+		readTS := readTimestamp.Timestamp()
 		exists, err := d.tableExistsAt(ctx, tableName, readTS)
 		if err != nil {
 			return err
@@ -199,10 +203,11 @@ func (d *DynamoDBServer) createTableWithRetry(ctx context.Context, tableName str
 		if err != nil {
 			return err
 		}
-		if _, err := d.coordinator.Dispatch(ctx, req); err == nil {
+		req.StartTS = readTS
+		dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
+		if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err == nil {
 			return nil
-		}
-		if !isRetryableTransactWriteError(err) {
+		} else if !isRetryableTransactWriteError(err) {
 			return errors.WithStack(err)
 		}
 		if err := waitRetryWithDeadline(ctx, deadline, backoff); err != nil {
@@ -293,7 +298,11 @@ func (d *DynamoDBServer) deleteTableWithRetry(ctx context.Context, tableName str
 	backoff := transactRetryInitialBackoff
 	deadline := time.Now().Add(transactRetryMaxDuration)
 	for range transactRetryMaxAttempts {
-		readTS := d.nextTxnReadTS()
+		readTimestamp, err := d.beginTxnReadTimestamp(ctx, "dynamodb delete table: begin read timestamp")
+		if err != nil {
+			return errors.WithStack(err)
+		}
+		readTS := readTimestamp.Timestamp()
 		schema, exists, err := d.loadTableSchemaAt(ctx, tableName, readTS)
 		if err != nil {
 			return errors.WithStack(err)
@@ -304,12 +313,13 @@ func (d *DynamoDBServer) deleteTableWithRetry(ctx context.Context, tableName str
 
 		req := &kv.OperationGroup[kv.OP]{
 			IsTxn:   true,
-			StartTS: 0,
+			StartTS: readTS,
 			Elems: []*kv.Elem[kv.OP]{
 				{Op: kv.Del, Key: dynamoTableMetaKey(tableName)},
 			},
 		}
-		if _, err := d.coordinator.Dispatch(ctx, req); err != nil {
+		dispatchCtx := readTimestamp.WithDispatchVoucher(ctx)
+		if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err != nil {
 			if !isRetryableTransactWriteError(err) {
 				return errors.WithStack(err)
 			}

adapter/dynamodb_transact.go

@@ -695,14 +695,15 @@ func (d *DynamoDBServer) transactWriteItemsWithRetry(ctx context.Context, in tra
 
 func (d *DynamoDBServer) runTransactWriteAttempt(
 	ctx context.Context,
-	reqs *kv.OperationGroup[kv.OP],
+	reqs *preparedTransactWriteItemsRequest,
 	generations map[string]uint64,
 	cleanupKeys [][]byte,
 ) (bool, error, error) {
-	if len(reqs.Elems) == 0 {
+	if len(reqs.group.Elems) == 0 {
 		return true, nil, nil
 	}
-	if _, err := d.coordinator.Dispatch(ctx, reqs); err != nil {
+	dispatchCtx := reqs.readTimestamp.WithDispatchVoucher(ctx)
+	if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, reqs.group); err != nil {
 		wrapped := errors.WithStack(err)
 		if !isRetryableTransactWriteError(err) {
 			return false, nil, wrapped
@@ -723,7 +724,12 @@ func (d *DynamoDBServer) runTransactWriteAttempt(
 	return false, nil, nil
 }
 
-func (d *DynamoDBServer) buildTransactWriteItemsRequest(ctx context.Context, in transactWriteItemsInput) (*kv.OperationGroup[kv.OP], map[string]uint64, [][]byte, error) {
+type preparedTransactWriteItemsRequest struct {
+	readTimestamp kv.ReadTimestamp
+	group         *kv.OperationGroup[kv.OP]
+}
+
+func (d *DynamoDBServer) buildTransactWriteItemsRequest(ctx context.Context, in transactWriteItemsInput) (*preparedTransactWriteItemsRequest, map[string]uint64, [][]byte, error) {
 	tableNames, err := collectTransactWriteTableNames(in)
 	if err != nil {
 		return nil, nil, nil, err
@@ -733,7 +739,11 @@ func (d *DynamoDBServer) buildTransactWriteItemsRequest(ctx context.Context, in
 			return nil, nil, nil, err
 		}
 	}
-	readTS := d.nextTxnReadTS()
+	readTimestamp, err := d.beginTxnReadTimestamp(ctx, "dynamodb transact-write: begin read timestamp")
+	if err != nil {
+		return nil, nil, nil, errors.WithStack(err)
+	}
+	readTS := readTimestamp.Timestamp()
 	reqs := &kv.OperationGroup[kv.OP]{
 		IsTxn: true,
 		// Keep transaction start aligned with the snapshot used to evaluate
@@ -752,7 +762,10 @@ func (d *DynamoDBServer) buildTransactWriteItemsRequest(ctx context.Context, in
 			return nil, nil, nil, err
 		}
 	}
-	return reqs, tableGenerations, cleanup, nil
+	return &preparedTransactWriteItemsRequest{
+		readTimestamp: readTimestamp,
+		group:         reqs,
+	}, tableGenerations, cleanup, nil
 }
 
 // processTransactWriteItem validates and plans a single item within a