feat(lease-read): leader-local lease read for coordinator and engine

adapter/distribution_server_test.go

@@ -747,3 +747,11 @@ func (s *distributionCoordinatorStub) Clock() *kv.HLC {
 func (s *distributionCoordinatorStub) LinearizableRead(_ context.Context) (uint64, error) {
 	return 0, nil
 }
+
+func (s *distributionCoordinatorStub) LeaseRead(ctx context.Context) (uint64, error) {
+	return s.LinearizableRead(ctx)
+}
+
+func (s *distributionCoordinatorStub) LeaseReadForKey(ctx context.Context, _ []byte) (uint64, error) {
+	return s.LinearizableRead(ctx)
+}

adapter/dynamodb.go

@@ -65,11 +65,17 @@ const (
 	transactRetryMaxBackoff     = 10 * time.Millisecond
 	transactRetryBackoffFactor  = 2
 	tableCleanupAsyncTimeout    = 5 * time.Minute
-	itemUpdateLockStripeCount   = 256
-	tableLockStripeCount        = 128
-	batchWriteItemMaxItems      = 25
-	transactGetItemsMaxItems    = 100
-	dynamoMaxRequestBodyBytes   = 1 << 20
+	// dynamoLeaseReadTimeout bounds how long LeaseReadForKey's slow
+	// path (LinearizableRead) may block before returning an error to
+	// the HTTP client. Matches the order of magnitude of Redis's
+	// redisDispatchTimeout so both adapters give up at similar
+	// wall-clock budgets on quorum loss.
+	dynamoLeaseReadTimeout    = 5 * time.Second
+	itemUpdateLockStripeCount = 256
+	tableLockStripeCount      = 128
+	batchWriteItemMaxItems    = 25
+	transactGetItemsMaxItems  = 100
+	dynamoMaxRequestBodyBytes = 1 << 20
 
 	dynamoTableMetaPrefix       = kv.DynamoTableMetaPrefix
 	dynamoTableGenerationPrefix = kv.DynamoTableGenerationPrefix
@@ -1340,38 +1346,85 @@ func (d *DynamoDBServer) commitItemWrite(ctx context.Context, req *kv.OperationG
 	return nil
 }
 
-func (d *DynamoDBServer) getItem(w http.ResponseWriter, r *http.Request) {
+func (d *DynamoDBServer) parseGetItemInput(w http.ResponseWriter, r *http.Request) (getItemInput, bool) {
 	body, err := io.ReadAll(maxDynamoBodyReader(w, r))
 	if err != nil {
 		writeDynamoError(w, http.StatusBadRequest, dynamoErrValidation, err.Error())
-		return
+		return getItemInput{}, false
 	}
 	var in getItemInput
 	if err := json.Unmarshal(body, &in); err != nil {
 		writeDynamoError(w, http.StatusBadRequest, dynamoErrValidation, err.Error())
-		return
+		return getItemInput{}, false
 	}
 	if strings.TrimSpace(in.TableName) == "" {
 		writeDynamoError(w, http.StatusBadRequest, dynamoErrValidation, "missing table name")
-		return
+		return getItemInput{}, false
 	}
 	if err := d.ensureLegacyTableMigration(r.Context(), in.TableName); err != nil {
 		writeDynamoErrorFromErr(w, err)
-		return
+		return getItemInput{}, false
 	}
+	return in, true
+}
 
-	readTS := d.resolveDynamoReadTS(in.ConsistentRead)
-	schema, exists, err := d.loadTableSchemaAt(r.Context(), in.TableName, readTS)
-	if err != nil {
+func (d *DynamoDBServer) getItem(w http.ResponseWriter, r *http.Request) {
+	in, ok := d.parseGetItemInput(w, r)
+	if !ok {
+		return
+	}
+	// Tentative TS for schema resolution only; schemas change rarely
+	// so a slight pre-lease stale is acceptable. The item read below
+	// is sampled AFTER the lease check.
+	tentativeTS := d.resolveDynamoReadTS(in.ConsistentRead)
+	_, itemKey, ok := d.resolveGetItemTarget(w, r, in, tentativeTS)
+	if !ok {
+		return
+	}
+	// Lease-check the shard that actually owns the ITEM key with a
+	// bounded timeout so a stalled Raft cannot hang this handler
+	// indefinitely if the client never cancels. Use defer so the
+	// cancel runs even if LeaseReadForKey panics or a future
+	// refactor inserts an early return; the cost of keeping ctx
+	// alive until handler exit is negligible because the next
+	// in-handler calls are local store reads.
+	leaseCtx, leaseCancel := context.WithTimeout(r.Context(), dynamoLeaseReadTimeout)
+	defer leaseCancel()
+	if _, err := kv.LeaseReadForKeyThrough(d.coordinator, leaseCtx, itemKey); err != nil {
 		writeDynamoError(w, http.StatusInternalServerError, dynamoErrInternal, err.Error())
 		return
 	}
-	if !exists {
-		writeDynamoError(w, http.StatusBadRequest, dynamoErrResourceNotFound, "table not found")
+	// Re-sample readTS AFTER the lease confirmation so that any write
+	// that completed on the same shard BEFORE the confirmation is
+	// visible. Sampling earlier would violate linearizability for
+	// ConsistentRead=false reads by returning a snapshot from before
+	// the most recent confirmed commit.
+	readTS := d.resolveDynamoReadTS(in.ConsistentRead)
+	// Pin readTS so concurrent MVCC GC cannot reclaim versions
+	// between the schema revalidation and the item read below;
+	// matches the pattern already used by queryItems / scanItems /
+	// transactGetItems.
+	readPin := d.pinReadTS(readTS)
+	defer readPin.Release()
+
+	// Re-resolve schema + itemKey at readTS and verify that the key
+	// we lease-checked is STILL the key that will be read. A table
+	// migration that commits between the tentative schema load and
+	// the lease confirmation may shift the item to a different shard
+	// even if the request parameters are unchanged, so comparing the
+	// computed item keys (not just generation) catches any future
+	// schema change that alters item routing.
+	finalSchema, freshItemKey, ok := d.resolveGetItemTarget(w, r, in, readTS)
+	if !ok {
+		return
+	}
+	if !bytes.Equal(freshItemKey, itemKey) {
+		writeDynamoError(w, http.StatusServiceUnavailable, dynamoErrInternal,
+			"table routing changed during read; please retry")
 		return
 	}
 
-	current, found, err := d.readLogicalItemAt(r.Context(), schema, in.Key, readTS)
+	current, found, err := d.readLogicalItemAt(r.Context(), finalSchema, in.Key, readTS)
 	if err != nil {
 		writeDynamoError(w, http.StatusBadRequest, dynamoErrValidation, err.Error())
 		return
@@ -1389,6 +1442,27 @@ func (d *DynamoDBServer) getItem(w http.ResponseWriter, r *http.Request) {
 	writeDynamoJSON(w, map[string]any{"Item": projected})
 }
 
+// resolveGetItemTarget loads the schema and computes the item key whose
+// shard must be lease-checked before the read. Returns false after
+// writing an error response; the caller should simply return.
+func (d *DynamoDBServer) resolveGetItemTarget(w http.ResponseWriter, r *http.Request, in getItemInput, readTS uint64) (*dynamoTableSchema, []byte, bool) {
+	schema, exists, err := d.loadTableSchemaAt(r.Context(), in.TableName, readTS)
+	if err != nil {
+		writeDynamoError(w, http.StatusInternalServerError, dynamoErrInternal, err.Error())
+		return nil, nil, false
+	}
+	if !exists {
+		writeDynamoError(w, http.StatusBadRequest, dynamoErrResourceNotFound, "table not found")
+		return nil, nil, false
+	}
+	itemKey, err := schema.itemKeyFromAttributes(in.Key)
+	if err != nil {
+		writeDynamoError(w, http.StatusBadRequest, dynamoErrValidation, err.Error())
+		return nil, nil, false
+	}
+	return schema, itemKey, true
+}
+
 func (d *DynamoDBServer) deleteItem(w http.ResponseWriter, r *http.Request) {
 	in, shouldReturnOld, err := decodeDeleteItemInput(maxDynamoBodyReader(w, r))
 	if err != nil {

adapter/dynamodb_test.go

@@ -1859,3 +1859,11 @@ func (w *testCoordinatorWrapper) Clock() *kv.HLC {
 func (w *testCoordinatorWrapper) LinearizableRead(ctx context.Context) (uint64, error) {
 	return w.inner.LinearizableRead(ctx)
 }
+
+func (w *testCoordinatorWrapper) LeaseRead(ctx context.Context) (uint64, error) {
+	return kv.LeaseReadThrough(w.inner, ctx)
+}
+
+func (w *testCoordinatorWrapper) LeaseReadForKey(ctx context.Context, key []byte) (uint64, error) {
+	return kv.LeaseReadForKeyThrough(w.inner, ctx, key)
+}