kv: persist prepare evidence across timestamp cutover

adapter/admin_grpc.go

@@ -763,6 +763,7 @@ var adminMethodPrefix = "/" + pb.Admin_ServiceDesc.ServiceName + "/"
 func adminTokenProtectedMethod(fullMethod string) bool {
 	return strings.HasPrefix(fullMethod, adminMethodPrefix) ||
 		fullMethod == pb.Internal_ForwardAdminProposal_FullMethodName ||
+		fullMethod == pb.Internal_GetPrepareEvidenceState_FullMethodName ||
 		fullMethod == pb.Internal_ForwardLeaseRead_FullMethodName
 }
 
@@ -790,24 +791,8 @@ func bearerTokenAuth(
 	if token == "" {
 		return nil, nil
 	}
-	expected := []byte(token)
 	check := func(ctx context.Context) error {
-		md, ok := metadata.FromIncomingContext(ctx)
-		if !ok {
-			return status.Error(codes.Unauthenticated, "missing authorization metadata")
-		}
-		values := md.Get("authorization")
-		if len(values) == 0 {
-			return status.Error(codes.Unauthenticated, "missing authorization header")
-		}
-		got, ok := strings.CutPrefix(values[0], "Bearer ")
-		if !ok {
-			return status.Error(codes.Unauthenticated, "authorization is not a bearer token")
-		}
-		if subtle.ConstantTimeCompare([]byte(got), expected) != 1 {
-			return status.Errorf(codes.Unauthenticated, "invalid %s token", credentialName)
-		}
-		return nil
+		return checkBearerToken(ctx, token, credentialName)
 	}
 	unary := func(
 		ctx context.Context,
@@ -840,6 +825,28 @@ func bearerTokenAuth(
 	return unary, stream
 }
 
+func checkBearerToken(ctx context.Context, token, credentialName string) error {
+	if token == "" {
+		return status.Errorf(codes.Unauthenticated, "%s credential is not configured", credentialName)
+	}
+	md, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		return errors.WithStack(status.Error(codes.Unauthenticated, "missing authorization metadata"))
+	}
+	values := md.Get("authorization")
+	if len(values) == 0 {
+		return errors.WithStack(status.Error(codes.Unauthenticated, "missing authorization header"))
+	}
+	got, ok := strings.CutPrefix(values[0], "Bearer ")
+	if !ok {
+		return errors.WithStack(status.Error(codes.Unauthenticated, "authorization is not a bearer token"))
+	}
+	if subtle.ConstantTimeCompare([]byte(got), []byte(token)) != 1 {
+		return status.Errorf(codes.Unauthenticated, "invalid %s token", credentialName)
+	}
+	return nil
+}
+
 // ErrAdminTokenRequired is returned by NewAdminServer helpers when the operator
 // failed to supply a token and also did not opt into insecure mode.
 var ErrAdminTokenRequired = errors.New("admin token file required; pass --adminInsecureNoAuth to run without")

adapter/distribution_server.go

@@ -15,6 +15,7 @@ import (
 	pb "github.com/bootjp/elastickv/proto"
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
+	"google.golang.org/genproto/googleapis/rpc/errdetails"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -319,6 +320,12 @@ func (s *DistributionServer) ValidateTimestamp(
 	if req == nil || req.GetTimestamp() == 0 {
 		return nil, errors.WithStack(status.Error(codes.InvalidArgument, "timestamp is required"))
 	}
+	if req.GetLegacyResolution() {
+		return s.validateLegacyResolutionTimestamp(ctx, req)
+	}
+	if req.GetStartTimestamp() != 0 {
+		return nil, errors.WithStack(status.Error(codes.InvalidArgument, "start timestamp requires legacy resolution mode"))
+	}
 	validator, ok := s.timestampAllocator.(kv.DurableTimestampValidator)
 	if !ok {
 		return nil, errors.WithStack(status.Error(codes.FailedPrecondition,
@@ -338,6 +345,20 @@ func (s *DistributionServer) ValidateTimestamp(
 	return resp, nil
 }
 
+func (s *DistributionServer) validateLegacyResolutionTimestamp(ctx context.Context, req *pb.ValidateTimestampRequest) (*pb.ValidateTimestampResponse, error) {
+	if req.GetStartTimestamp() == 0 || req.GetStartTimestamp() >= req.GetTimestamp() {
+		return nil, errors.WithStack(status.Error(codes.InvalidArgument, "legacy resolution requires 0 < start < commit"))
+	}
+	validator, ok := s.timestampAllocator.(kv.DurableLegacyResolutionValidator)
+	if !ok {
+		return nil, errors.WithStack(status.Error(codes.FailedPrecondition, "pending legacy resolution validation is not configured"))
+	}
+	if err := validator.ValidateLegacyResolutionTimestamps(ctx, req.GetStartTimestamp(), req.GetTimestamp()); err != nil {
+		return nil, timestampRPCError(err)
+	}
+	return &pb.ValidateTimestampResponse{Valid: true, LegacyResolutionValidated: true}, nil
+}
+
 func (s *DistributionServer) allocateTimestampReservation(
 	ctx context.Context,
 	count int,
@@ -407,6 +428,9 @@ func timestampRPCError(err error) error {
 	if code := status.Code(err); code != codes.Unknown {
 		return err
 	}
+	if mapped := phaseDActivationRPCError(err); mapped != nil {
+		return mapped
+	}
 	switch {
 	case errors.Is(err, context.Canceled):
 		return errors.WithStack(status.Error(codes.Canceled, err.Error()))
@@ -420,11 +444,33 @@ func timestampRPCError(err error) error {
 		return errors.WithStack(status.Error(codes.OutOfRange, err.Error()))
 	case errors.Is(err, kv.ErrTSOTimestampInvalid):
 		return errors.WithStack(status.Error(codes.InvalidArgument, err.Error()))
+	default:
+		return errors.WithStack(status.Error(codes.Unavailable, err.Error()))
+	}
+}
+
+func phaseDActivationRPCError(err error) error {
+	switch {
 	case errors.Is(err, kv.ErrTSOPhaseDInactive):
+		return inactiveTimestampRPCError(err)
+	case errors.Is(err, kv.ErrTSOPhaseDActivationPending):
+		return errors.WithStack(status.Error(codes.ResourceExhausted, err.Error()))
+	case errors.Is(err, kv.ErrTSOPhaseDActivationGateRequired):
 		return errors.WithStack(status.Error(codes.FailedPrecondition, err.Error()))
 	default:
-		return errors.WithStack(status.Error(codes.Unavailable, err.Error()))
+		return nil
+	}
+}
+
+func inactiveTimestampRPCError(err error) error {
+	s := status.New(codes.FailedPrecondition, err.Error())
+	withDetail, detailErr := s.WithDetails(&errdetails.ErrorInfo{
+		Domain: kv.TSOValidationErrorDomain, Reason: kv.TSOPhaseDInactiveReason,
+	})
+	if detailErr != nil {
+		return errors.WithStack(s.Err())
 	}
+	return errors.WithStack(withDetail.Err())
 }
 
 // ListRoutes returns all durable routes from catalog storage.

adapter/distribution_server_test.go

@@ -261,6 +261,80 @@ func TestDistributionServerValidateTimestamp(t *testing.T) {
 	require.Equal(t, codes.OutOfRange, status.Code(err))
 }
 
+type pendingResolutionAllocator struct {
+	*distributionTSOAllocator
+	pendingFloor uint64
+}
+
+func (a *pendingResolutionAllocator) ValidateDurableTimestamp(ctx context.Context, timestamp uint64) error {
+	if !a.phaseD {
+		return kv.ErrTSOPhaseDInactive
+	}
+	return a.distributionTSOAllocator.ValidateDurableTimestamp(ctx, timestamp)
+}
+
+func (a *pendingResolutionAllocator) ValidateLegacyResolutionTimestamps(_ context.Context, start, commit uint64) error {
+	if a.phaseD {
+		return kv.ErrTSOPhaseDInactive
+	}
+	if start == 0 || start >= commit || (a.pendingFloor != 0 && commit > a.pendingFloor) {
+		return kv.ErrTSOTimestampInvalid
+	}
+	return nil
+}
+
+func TestDistributionServerPendingLegacyResolutionMode(t *testing.T) {
+	alloc := &pendingResolutionAllocator{distributionTSOAllocator: &distributionTSOAllocator{leader: true}, pendingFloor: 100}
+	s := NewDistributionServer(distribution.NewEngine(), nil, WithDistributionTimestampAllocator(alloc))
+	ctx := context.Background()
+	resp, err := s.ValidateTimestamp(ctx, &pb.ValidateTimestampRequest{Timestamp: 100, StartTimestamp: 50, LegacyResolution: true})
+	require.NoError(t, err)
+	require.True(t, resp.GetValid())
+	require.True(t, resp.GetLegacyResolutionValidated())
+	require.False(t, resp.GetPhaseDActive())
+	for _, req := range []*pb.ValidateTimestampRequest{
+		{Timestamp: 100, LegacyResolution: true},
+		{Timestamp: 100, StartTimestamp: 100, LegacyResolution: true},
+		{Timestamp: 101, StartTimestamp: 50, LegacyResolution: true},
+		{Timestamp: 100, StartTimestamp: 50},
+	} {
+		_, err := s.ValidateTimestamp(ctx, req)
+		require.Equal(t, codes.InvalidArgument, status.Code(err))
+	}
+	_, err = s.ValidateTimestamp(ctx, &pb.ValidateTimestampRequest{Timestamp: 100})
+	require.Error(t, err, "ordinary validation must remain closed before M7")
+	alloc.pendingFloor = 0
+	resp, err = s.ValidateTimestamp(ctx, &pb.ValidateTimestampRequest{Timestamp: 101, StartTimestamp: 50, LegacyResolution: true})
+	require.NoError(t, err, "pre-pending drain must admit existing resolutions above the future barrier")
+	require.True(t, resp.GetLegacyResolutionValidated())
+	alloc.phaseD = true
+	_, err = s.ValidateTimestamp(ctx, &pb.ValidateTimestampRequest{Timestamp: 100, StartTimestamp: 50, LegacyResolution: true})
+	require.Equal(t, codes.FailedPrecondition, status.Code(err))
+}
+
+func TestDistributionServerPendingLegacyResolutionRoutedWrappers(t *testing.T) {
+	alloc := &pendingResolutionAllocator{distributionTSOAllocator: &distributionTSOAllocator{leader: true}, pendingFloor: 100}
+	addr := serveDistributionTestServer(t, NewDistributionServer(distribution.NewEngine(), nil, WithDistributionTimestampAllocator(alloc)))
+	routed, err := kv.NewLeaderRoutedTSOAllocator(&distributionTSOAllocator{}, distributionLeaderView{addr: addr})
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, routed.Close()) })
+	batch, err := kv.NewBatchAllocator(routed, 2)
+	require.NoError(t, err)
+	dynamic := kv.NewDynamicTimestampAllocator(batch)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	require.ErrorIs(t, dynamic.ValidateDurableTimestamp(ctx, 100), kv.ErrTSOPhaseDInactive)
+	require.NoError(t, dynamic.ValidateLegacyResolutionTimestamps(ctx, 50, 100))
+	require.NoError(t, ctx.Err(), "inactive must not consume the retry budget before resolution")
+	require.Error(t, batch.ValidateLegacyResolutionTimestamps(context.Background(), 50, 101))
+}
+
+func TestDistributionServerPendingLegacyResolutionRejectsUnsupportedAllocator(t *testing.T) {
+	s := NewDistributionServer(distribution.NewEngine(), nil, WithDistributionTimestampAllocator(&distributionTSOAllocator{}))
+	_, err := s.ValidateTimestamp(context.Background(), &pb.ValidateTimestampRequest{Timestamp: 100, StartTimestamp: 50, LegacyResolution: true})
+	require.Equal(t, codes.FailedPrecondition, status.Code(err))
+}
+
 func TestDistributionServerValidateTimestamp_LeaderRoutedRPC(t *testing.T) {
 	serverAlloc := &distributionTSOAllocator{
 		base:        701,
@@ -402,7 +476,7 @@ func TestDistributionServerGetTimestamp_LeaderRoutedActivatesCutover(t *testing.
 	require.True(t, serverAlloc.activate)
 }
 
-func serveDistributionTestServer(t *testing.T, server *DistributionServer) string {
+func serveDistributionTestServer(t *testing.T, server pb.DistributionServer) string {
 	t.Helper()
 	listener, err := new(net.ListenConfig).Listen(context.Background(), "tcp", "127.0.0.1:0")
 	require.NoError(t, err)
@@ -421,6 +495,22 @@ func serveDistributionTestServer(t *testing.T, server *DistributionServer) strin
 	return listener.Addr().String()
 }
 
+type legacyIgnoringTimestampValidationServer struct {
+	pb.UnimplementedDistributionServer
+}
+
+func (legacyIgnoringTimestampValidationServer) ValidateTimestamp(context.Context, *pb.ValidateTimestampRequest) (*pb.ValidateTimestampResponse, error) {
+	return &pb.ValidateTimestampResponse{Valid: true, PhaseDActive: true}, nil
+}
+
+func TestDistributionServerPendingLegacyResolutionRejectsOldModeIgnoringServer(t *testing.T) {
+	addr := serveDistributionTestServer(t, legacyIgnoringTimestampValidationServer{})
+	routed, err := kv.NewLeaderRoutedTSOAllocator(&distributionTSOAllocator{}, distributionLeaderView{addr: addr})
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, routed.Close()) })
+	require.ErrorIs(t, routed.ValidateLegacyResolutionTimestamps(context.Background(), 50, 100), kv.ErrTSOProtocolUnsupported)
+}
+
 func TestNewDistributionServer_DefaultCatalogReloadRetryPolicy(t *testing.T) {
 	t.Parallel()
 

adapter/internal.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"os"
 	"strings"
+	"sync/atomic"
 
 	"github.com/bootjp/elastickv/distribution"
 	"github.com/bootjp/elastickv/internal"
@@ -55,6 +56,16 @@ func WithInternalStore(st store.MVCCStore) InternalOption {
 	}
 }
 
+func WithInternalPrimaryCommitEvidence(reader kv.PrimaryCommitEvidenceReader) InternalOption {
+	return func(i *Internal) {
+		i.primaryCommitEvidence = reader
+	}
+}
+
+func WithInternalTSOCutoverState(state interface{ CutoverActive() bool }) InternalOption {
+	return func(i *Internal) { i.tsoCutoverState = state }
+}
+
 func WithInternalMigrationProposer(proposer raftengine.Proposer) InternalOption {
 	return func(i *Internal) {
 		i.migrationProposer = proposer
@@ -127,6 +138,10 @@ type Internal struct {
 	lastCommitTimestamp     func() uint64
 	relay                   *RedisPubSubRelay
 	store                   store.MVCCStore
+	primaryCommitEvidence   kv.PrimaryCommitEvidenceReader
+	tsoCutoverState         interface{ CutoverActive() bool }
+	prepareEvidenceToken    string
+	prepareEvidenceScanning atomic.Bool
 	migrationProposer       raftengine.Proposer
 	migrationImportGate     func(context.Context) error
 	migrationPromoteGate    func(context.Context) error
@@ -192,6 +207,9 @@ func (i *Internal) Forward(ctx context.Context, req *pb.ForwardRequest) (*pb.For
 			CommitIndex: 0,
 		}, errors.WithStack(err)
 	}
+	if err := kv.StampTimestampAdmission(ctx, i.tsAllocator, i.store, req.GetRequests()); err != nil {
+		return nil, errors.WithStack(err)
+	}
 
 	r, err := i.transactionManager.Commit(ctx, req.Requests)
 	if err != nil {
@@ -998,6 +1016,9 @@ func (i *Internal) stampTxnTimestamps(ctx context.Context, reqs []*pb.Request) (
 		}
 	}
 
+	if err := i.validateForwardedPrepareEvidence(ctx, reqs, startTS); err != nil {
+		return 0, err
+	}
 	commitTS, err := i.fillForwardedTxnCommitTS(ctx, reqs, startTS)
 	if err != nil {
 		return 0, err
@@ -1010,9 +1031,19 @@ func (i *Internal) stampTxnTimestamps(ctx context.Context, reqs []*pb.Request) (
 
 func (i *Internal) validateForwardedTxnStartForPhase(ctx context.Context, reqs []*pb.Request, startTS uint64) error {
 	const label = "stampTxnTimestamps: forwarded start ts"
-	if forwardedTxnResolutionOnly(reqs) {
+	if forwardedTxnCreationsHavePrepareProof(reqs) {
 		err := kv.ValidateDurablePersistenceTimestamp(ctx, i.tsAllocator, startTS, label)
 		if errors.Is(err, kv.ErrTSOTimestampPrePhaseD) {
+			// Applied-read vouchers can legitimately retain an old startTS.
+			// The chosen commit proof and current admission epoch are validated
+			// before proposing; no legacy creation-window exemption is needed.
+			return nil
+		}
+		return errors.WithStack(err)
+	}
+	if forwardedTxnResolutionOnly(reqs) {
+		err := kv.ValidateDurablePersistenceTimestamp(ctx, i.tsAllocator, startTS, label)
+		if errors.Is(err, kv.ErrTSOTimestampPrePhaseD) || errors.Is(err, kv.ErrTSOPhaseDInactive) {
 			return nil // Existing intents may need COMMIT or ABORT after the creation window closes.
 		}
 		return errors.WithStack(err)
@@ -1111,7 +1142,9 @@ func (i *Internal) fillForwardedTxnCommitTS(ctx context.Context, reqs []*pb.Requ
 		}
 	} else if err := kv.ValidateForwardedTxnCommitTimestamp(
 		ctx, i.tsAllocator, startTS, commitTS, resolution,
-		"fillForwardedTxnCommitTS: forwarded commit ts"); err != nil {
+		"fillForwardedTxnCommitTS: forwarded commit ts", func(ctx context.Context, pending bool) error {
+			return i.verifyCommitEvidence(ctx, reqs, startTS, commitTS, pending)
+		}); err != nil {
 		// A commit timestamp that arrived already set in the transaction meta
 		// is the timestamp every mutation in this batch is persisted under, and
 		// it never passed through the coordinator's own validation. A legacy