diff --git a/adapter/distribution_server.go b/adapter/distribution_server.go index d0abb3fe5..297c8985b 100644 --- a/adapter/distribution_server.go +++ b/adapter/distribution_server.go @@ -153,7 +153,7 @@ func (s *DistributionServer) SplitRange(ctx context.Context, req *pb.SplitRangeR return nil, err } - parent, _, found := findRouteByID(snapshot.Routes, req.GetRouteId()) + parent, found := findRouteByID(snapshot.Routes, req.GetRouteId()) if !found { return nil, grpcStatusError(codes.NotFound, errDistributionUnknownRoute.Error()) } @@ -167,7 +167,7 @@ func (s *DistributionServer) SplitRange(ctx context.Context, req *pb.SplitRangeR if err != nil { return nil, err } - left, right := splitCatalogRoutes(parent, splitKey, leftID, rightID) + left, right := splitCatalogRoutes(parent, splitKey, leftID, rightID, 0) saved, err := s.saveSplitResultViaCoordinator(ctx, snapshot.ReadTS, req.GetExpectedCatalogVersion(), parent.RouteID, left, right) if err != nil { @@ -176,11 +176,15 @@ func (s *DistributionServer) SplitRange(ctx context.Context, req *pb.SplitRangeR if err := s.applyEngineSnapshot(saved); err != nil { return nil, err } + savedLeft, savedRight, err := splitChildrenFromSnapshot(saved, left.RouteID, right.RouteID) + if err != nil { + return nil, err + } return &pb.SplitRangeResponse{ CatalogVersion: saved.Version, - Left: toProtoRouteDescriptor(left), - Right: toProtoRouteDescriptor(right), + Left: toProtoRouteDescriptor(savedLeft), + Right: toProtoRouteDescriptor(savedRight), }, nil } @@ -226,14 +230,18 @@ func (s *DistributionServer) saveSplitResultViaCoordinator( if err != nil { return distribution.CatalogSnapshot{}, grpcStatusErrorf(codes.Internal, "build split mutations: %v", err) } - if _, err := s.coordinator.Dispatch(ctx, &kv.OperationGroup[kv.OP]{ + resp, err := s.coordinator.Dispatch(ctx, &kv.OperationGroup[kv.OP]{ Elems: ops, IsTxn: true, StartTS: readTS, - }); err != nil { + }) + if err != nil { return distribution.CatalogSnapshot{}, grpcStatusErrorf(codes.Internal, "commit split mutations: %v", err) } - return s.loadCatalogSnapshotAtLeastVersion(ctx, nextVersion) + if resp == nil || resp.CommitTS == 0 { + return distribution.CatalogSnapshot{}, grpcStatusError(codes.Internal, "split commit timestamp missing") + } + return s.loadCatalogSnapshotAtVersion(ctx, resp.CommitTS, nextVersion) } func buildCatalogSplitOps( @@ -255,10 +263,15 @@ func buildCatalogSplitOps( if err != nil { return nil, errors.WithStack(err) } + patchOffset, err := splitAtHLCPatchOffset(encoded) + if err != nil { + return nil, err + } ops = append(ops, &kv.Elem[kv.OP]{ - Op: kv.Put, - Key: distribution.CatalogRouteKey(route.RouteID), - Value: encoded, + Op: kv.Put, + Key: distribution.CatalogRouteKey(route.RouteID), + Value: encoded, + CommitTSValueOffset: patchOffset, }) } ops = append(ops, &kv.Elem[kv.OP]{ @@ -274,6 +287,26 @@ func buildCatalogSplitOps( return ops, nil } +func splitAtHLCPatchOffset(encoded []byte) (uint64, error) { + const splitAtHLCTailBytes = 8 + if len(encoded) < splitAtHLCTailBytes { + return 0, errors.WithStack(distribution.ErrCatalogInvalidRouteRecord) + } + return uint64(len(encoded) - splitAtHLCTailBytes), nil //nolint:gosec // len was checked to be at least splitAtHLCTailBytes. +} + +func splitChildrenFromSnapshot(snapshot distribution.CatalogSnapshot, leftID uint64, rightID uint64) (distribution.RouteDescriptor, distribution.RouteDescriptor, error) { + left, found := findRouteByID(snapshot.Routes, leftID) + if !found { + return distribution.RouteDescriptor{}, distribution.RouteDescriptor{}, grpcStatusError(codes.Internal, "catalog split committed but left child is missing") + } + right, found := findRouteByID(snapshot.Routes, rightID) + if !found { + return distribution.RouteDescriptor{}, distribution.RouteDescriptor{}, grpcStatusError(codes.Internal, "catalog split committed but right child is missing") + } + return left, right, nil +} + func (s *DistributionServer) loadCatalogSnapshot(ctx context.Context) (distribution.CatalogSnapshot, error) { if s.catalog == nil { return distribution.CatalogSnapshot{}, grpcStatusError(codes.FailedPrecondition, errDistributionCatalogNotConfigured.Error()) @@ -285,9 +318,10 @@ func (s *DistributionServer) loadCatalogSnapshot(ctx context.Context) (distribut return snapshot, nil } -func (s *DistributionServer) loadCatalogSnapshotAtLeastVersion( +func (s *DistributionServer) loadCatalogSnapshotAtVersion( ctx context.Context, - minVersion uint64, + readTS uint64, + wantVersion uint64, ) (distribution.CatalogSnapshot, error) { attempts := s.reloadRetry.attempts if attempts <= 0 { @@ -300,11 +334,11 @@ func (s *DistributionServer) loadCatalogSnapshotAtLeastVersion( var last distribution.CatalogSnapshot for attempt := 0; attempt < attempts; attempt++ { - snapshot, err := s.catalog.Snapshot(ctx) + snapshot, err := s.catalog.SnapshotAt(ctx, readTS) if err != nil { return distribution.CatalogSnapshot{}, grpcStatusErrorf(codes.Internal, "reload route catalog: %v", err) } - if snapshot.Version >= minVersion { + if snapshot.Version == wantVersion { return snapshot, nil } last = snapshot @@ -317,9 +351,10 @@ func (s *DistributionServer) loadCatalogSnapshotAtLeastVersion( } return distribution.CatalogSnapshot{}, grpcStatusErrorf( codes.Internal, - "catalog split committed but local snapshot is stale: got %d, want at least %d", + "catalog split committed but local snapshot is stale at %d: got %d, want %d", + readTS, last.Version, - minVersion, + wantVersion, ) } @@ -381,6 +416,7 @@ func splitCatalogRoutes( splitKey []byte, leftID uint64, rightID uint64, + splitAtHLC uint64, ) (distribution.RouteDescriptor, distribution.RouteDescriptor) { // parent and splitKey are already cloned before this point and are immutable here. left := distribution.RouteDescriptor{ @@ -390,6 +426,7 @@ func splitCatalogRoutes( GroupID: parent.GroupID, State: parent.State, ParentRouteID: parent.RouteID, + SplitAtHLC: splitAtHLC, } right := distribution.RouteDescriptor{ RouteID: rightID, @@ -398,6 +435,7 @@ func splitCatalogRoutes( GroupID: parent.GroupID, State: parent.State, ParentRouteID: parent.RouteID, + SplitAtHLC: splitAtHLC, } return left, right } @@ -428,13 +466,13 @@ func (s *DistributionServer) allocateChildRouteIDs(ctx context.Context, readTS u return leftID, rightID, nil } -func findRouteByID(routes []distribution.RouteDescriptor, routeID uint64) (distribution.RouteDescriptor, int, bool) { - for i, route := range routes { +func findRouteByID(routes []distribution.RouteDescriptor, routeID uint64) (distribution.RouteDescriptor, bool) { + for _, route := range routes { if route.RouteID == routeID { - return distribution.CloneRouteDescriptor(route), i, true + return distribution.CloneRouteDescriptor(route), true } } - return distribution.RouteDescriptor{}, -1, false + return distribution.RouteDescriptor{}, false } func toProtoRouteDescriptors(routes []distribution.RouteDescriptor) []*pb.RouteDescriptor { @@ -453,6 +491,7 @@ func toProtoRouteDescriptor(route distribution.RouteDescriptor) *pb.RouteDescrip RaftGroupId: route.GroupID, State: toProtoRouteState(route.State), ParentRouteId: route.ParentRouteID, + SplitAtHlc: route.SplitAtHLC, } } diff --git a/adapter/distribution_server_test.go b/adapter/distribution_server_test.go index 030116c34..68401bede 100644 --- a/adapter/distribution_server_test.go +++ b/adapter/distribution_server_test.go @@ -2,6 +2,7 @@ package adapter import ( "context" + "encoding/binary" "testing" "time" @@ -85,6 +86,7 @@ func TestDistributionServerListRoutes_ReadsDurableCatalog(t *testing.T) { GroupID: 2, State: distribution.RouteStateWriteFenced, ParentRouteID: 1, + SplitAtHLC: 99, }, { RouteID: 1, @@ -111,6 +113,7 @@ func TestDistributionServerListRoutes_ReadsDurableCatalog(t *testing.T) { require.Equal(t, uint64(2), resp.Routes[1].RouteId) require.Nil(t, resp.Routes[1].End) require.Equal(t, pb.RouteState_ROUTE_STATE_WRITE_FENCED, resp.Routes[1].State) + require.Equal(t, uint64(99), resp.Routes[1].SplitAtHlc) } func TestDistributionServerListRoutes_RequiresCatalog(t *testing.T) { @@ -174,6 +177,8 @@ func TestDistributionServerSplitRange_Success(t *testing.T) { require.Equal(t, []byte("m"), resp.Right.End) require.Equal(t, uint64(1), resp.Right.RaftGroupId) require.Equal(t, uint64(1), resp.Right.ParentRouteId) + require.NotZero(t, resp.Left.SplitAtHlc) + require.Equal(t, resp.Left.SplitAtHlc, resp.Right.SplitAtHlc) snapshot, err := catalog.Snapshot(ctx) require.NoError(t, err) @@ -183,6 +188,9 @@ func TestDistributionServerSplitRange_Success(t *testing.T) { require.Equal(t, uint64(3), snapshot.Routes[0].RouteID) require.Equal(t, uint64(4), snapshot.Routes[1].RouteID) require.Equal(t, uint64(2), snapshot.Routes[2].RouteID) + require.NotZero(t, snapshot.Routes[0].SplitAtHLC) + require.Equal(t, snapshot.Routes[0].SplitAtHLC, snapshot.Routes[1].SplitAtHLC) + require.Equal(t, snapshot.Routes[0].SplitAtHLC, resp.Left.SplitAtHlc) require.Equal(t, uint64(2), engine.Version()) leftRoute, ok := engine.GetRoute([]byte("b")) @@ -304,6 +312,20 @@ func TestDistributionServerSplitRange_UsesCoordinatorForCatalogWrites(t *testing require.Equal(t, uint64(2), resp.CatalogVersion) require.Equal(t, 1, coordinator.dispatchCalls) require.Equal(t, readSnapshot.ReadTS, coordinator.lastStartTS) + require.Zero(t, coordinator.lastRequestedCommitTS) + require.NotZero(t, coordinator.lastCommitTS) + require.Greater(t, coordinator.lastCommitTS, coordinator.lastStartTS) + + snapshot, err := catalog.Snapshot(ctx) + require.NoError(t, err) + left, found := findRouteByID(snapshot.Routes, resp.Left.RouteId) + require.True(t, found) + right, found := findRouteByID(snapshot.Routes, resp.Right.RouteId) + require.True(t, found) + require.Equal(t, coordinator.lastCommitTS, left.SplitAtHLC) + require.Equal(t, coordinator.lastCommitTS, right.SplitAtHLC) + require.Equal(t, coordinator.lastCommitTS, resp.Left.SplitAtHlc) + require.Equal(t, coordinator.lastCommitTS, resp.Right.SplitAtHlc) } func TestDistributionServerSplitRange_UsesPersistentNextRouteID(t *testing.T) { @@ -380,7 +402,7 @@ func TestDistributionServerSplitRange_UsesPersistentNextRouteID(t *testing.T) { require.Equal(t, uint64(6), second.Right.RouteId) } -func TestDistributionServerSplitRange_AllowsVersionAdvanceAfterCommit(t *testing.T) { +func TestDistributionServerSplitRange_ReturnsExactCommittedSplitVersion(t *testing.T) { t.Parallel() ctx := context.Background() @@ -419,8 +441,20 @@ func TestDistributionServerSplitRange_AllowsVersionAdvanceAfterCommit(t *testing SplitKey: []byte("g"), }) require.NoError(t, err) - require.Equal(t, uint64(3), resp.CatalogVersion) - require.Equal(t, uint64(3), engine.Version()) + require.Equal(t, uint64(2), resp.CatalogVersion) + require.Equal(t, uint64(2), engine.Version()) + require.Equal(t, uint64(3), resp.Left.RouteId) + require.Equal(t, uint64(4), resp.Right.RouteId) + require.Equal(t, coordinator.lastCommitTS, resp.Left.SplitAtHlc) + require.Equal(t, coordinator.lastCommitTS, resp.Right.SplitAtHlc) + + committed, err := catalog.SnapshotAt(ctx, coordinator.lastCommitTS) + require.NoError(t, err) + require.Equal(t, uint64(2), committed.Version) + + latest, err := catalog.Snapshot(ctx) + require.NoError(t, err) + require.Equal(t, uint64(3), latest.Version) } func TestDistributionServerSplitRange_RetriesCatalogReloadUntilVisible(t *testing.T) { @@ -588,20 +622,24 @@ func seededDistributionServerWithoutCoordinator(t *testing.T) (*DistributionServ } type distributionCoordinatorStub struct { - store store.MVCCStore - leader bool - nextTS uint64 - lastStartTS uint64 - afterDispatch func(context.Context, store.MVCCStore, uint64) error - asyncApplyDone chan error - asyncApplyDelay time.Duration - dispatchCalls int + store store.MVCCStore + leader bool + clock *kv.HLC + nextTS uint64 + lastStartTS uint64 + lastCommitTS uint64 + lastRequestedCommitTS uint64 + afterDispatch func(context.Context, store.MVCCStore, uint64) error + asyncApplyDone chan error + asyncApplyDelay time.Duration + dispatchCalls int } func newDistributionCoordinatorStub(st store.MVCCStore, leader bool) *distributionCoordinatorStub { return &distributionCoordinatorStub{ store: st, leader: leader, + clock: kv.NewHLC(), } } @@ -610,10 +648,15 @@ func (s *distributionCoordinatorStub) Dispatch(ctx context.Context, reqs *kv.Ope return nil, err } s.dispatchCalls++ - startTS, commitTS := s.nextTimestamps(reqs.StartTS) + s.lastRequestedCommitTS = reqs.CommitTS + startTS, commitTS := s.nextTimestamps(reqs.StartTS, reqs.CommitTS) s.lastStartTS = startTS + s.lastCommitTS = commitTS - mutations, err := coordinatorStubMutations(reqs.Elems) + if err := kv.ValidateElemCommitTSPatches(reqs.Elems, commitTS); err != nil { + return nil, err + } + mutations, err := coordinatorStubMutations(reqs.Elems, commitTS) if err != nil { return nil, err } @@ -627,12 +670,12 @@ func (s *distributionCoordinatorStub) Dispatch(ctx context.Context, reqs *kv.Ope done <- err } }() - return &kv.CoordinateResponse{CommitIndex: commitTS}, nil + return &kv.CoordinateResponse{CommitIndex: commitTS, CommitTS: commitTS}, nil } if err := s.applyDispatch(ctx, mutations, startTS, commitTS); err != nil { return nil, err } - return &kv.CoordinateResponse{CommitIndex: commitTS}, nil + return &kv.CoordinateResponse{CommitIndex: commitTS, CommitTS: commitTS}, nil } func (s *distributionCoordinatorStub) validateDispatch(reqs *kv.OperationGroup[kv.OP]) error { @@ -645,7 +688,13 @@ func (s *distributionCoordinatorStub) validateDispatch(reqs *kv.OperationGroup[k return nil } -func (s *distributionCoordinatorStub) nextTimestamps(startTS uint64) (uint64, uint64) { +func (s *distributionCoordinatorStub) nextTimestamps(startTS uint64, requestedCommitTS uint64) (uint64, uint64) { + if requestedCommitTS != 0 { + if s.clock != nil { + s.clock.Observe(requestedCommitTS) + } + return startTS, requestedCommitTS + } if s.nextTS == 0 { s.nextTS = s.store.LastCommitTS() + 1 } @@ -674,10 +723,10 @@ func (s *distributionCoordinatorStub) applyDispatch( return nil } -func coordinatorStubMutations(elems []*kv.Elem[kv.OP]) ([]*store.KVPairMutation, error) { +func coordinatorStubMutations(elems []*kv.Elem[kv.OP], commitTS uint64) ([]*store.KVPairMutation, error) { mutations := make([]*store.KVPairMutation, 0, len(elems)) for _, elem := range elems { - mutation, err := coordinatorStubMutation(elem) + mutation, err := coordinatorStubMutation(elem, commitTS) if err != nil { return nil, err } @@ -686,16 +735,20 @@ func coordinatorStubMutations(elems []*kv.Elem[kv.OP]) ([]*store.KVPairMutation, return mutations, nil } -func coordinatorStubMutation(elem *kv.Elem[kv.OP]) (*store.KVPairMutation, error) { +func coordinatorStubMutation(elem *kv.Elem[kv.OP], commitTS uint64) (*store.KVPairMutation, error) { if elem == nil { return nil, kv.ErrInvalidRequest } switch elem.Op { case kv.Put: + value := distribution.CloneBytes(elem.Value) + if elem.CommitTSValueOffset != 0 { + binary.BigEndian.PutUint64(value[elem.CommitTSValueOffset:elem.CommitTSValueOffset+8], commitTS) + } return &store.KVPairMutation{ Op: store.OpTypePut, Key: distribution.CloneBytes(elem.Key), - Value: distribution.CloneBytes(elem.Value), + Value: value, }, nil case kv.Del: return &store.KVPairMutation{ @@ -740,7 +793,7 @@ func (s *distributionCoordinatorStub) RaftLeaderForKey(_ []byte) string { } func (s *distributionCoordinatorStub) Clock() *kv.HLC { - return nil + return s.clock } func (s *distributionCoordinatorStub) LinearizableRead(_ context.Context) (uint64, error) { diff --git a/adapter/internal.go b/adapter/internal.go index aa98f16b9..bf4279e31 100644 --- a/adapter/internal.go +++ b/adapter/internal.go @@ -55,7 +55,8 @@ func (i *Internal) Forward(ctx context.Context, req *pb.ForwardRequest) (*pb.For return nil, errors.WithStack(ErrNotLeader) } - if err := i.stampTimestamps(ctx, req); err != nil { + commitTS, err := i.stampTimestamps(ctx, req) + if err != nil { return &pb.ForwardResponse{ Success: false, CommitIndex: 0, @@ -73,6 +74,7 @@ func (i *Internal) Forward(ctx context.Context, req *pb.ForwardRequest) (*pb.For return &pb.ForwardResponse{ Success: true, CommitIndex: r.CommitIndex, + CommitTs: commitTS, }, nil } @@ -85,15 +87,15 @@ func (i *Internal) RelayPublish(_ context.Context, req *pb.RelayPublishRequest) }, nil } -func (i *Internal) stampTimestamps(ctx context.Context, req *pb.ForwardRequest) error { +func (i *Internal) stampTimestamps(ctx context.Context, req *pb.ForwardRequest) (uint64, error) { if req == nil { - return nil + return 0, nil } if req.IsTxn { return i.stampTxnTimestamps(ctx, req.Requests) } - return i.stampRawTimestamps(ctx, req.Requests) + return 0, i.stampRawTimestamps(ctx, req.Requests) } func (i *Internal) nextTimestamp(ctx context.Context, label string) (uint64, error) { @@ -172,17 +174,17 @@ func (i *Internal) stampRawTimestamps(ctx context.Context, reqs []*pb.Request) e return nil } -func (i *Internal) stampTxnTimestamps(ctx context.Context, reqs []*pb.Request) error { +func (i *Internal) stampTxnTimestamps(ctx context.Context, reqs []*pb.Request) (uint64, error) { startTS := forwardedTxnStartTS(reqs) if startTS == 0 { ts, err := i.nextTimestamp(ctx, "stampTxnTimestamps startTS") if err != nil { - return err + return 0, err } startTS = ts } if startTS == ^uint64(0) { - return errors.WithStack(ErrTxnTimestampOverflow) + return 0, errors.WithStack(ErrTxnTimestampOverflow) } // Assign the unified timestamp to all requests in the transaction. @@ -220,13 +222,49 @@ func forwardedTxnMetaMutation(r *pb.Request, metaPrefix []byte) (*pb.Mutation, b return r.Mutations[0], true } -func (i *Internal) fillForwardedTxnCommitTS(ctx context.Context, reqs []*pb.Request, startTS uint64) error { - type metaToUpdate struct { - m *pb.Mutation - meta kv.TxnMeta +type forwardedTxnMetaToUpdate struct { + m *pb.Mutation + meta kv.TxnMeta +} + +func (i *Internal) fillForwardedTxnCommitTS(ctx context.Context, reqs []*pb.Request, startTS uint64) (uint64, error) { + metaMutations, commitTS, err := collectForwardedTxnMetas(reqs) + if err != nil { + return 0, err } - metaMutations := make([]metaToUpdate, 0, len(reqs)) + if commitTS == 0 && len(metaMutations) > 0 { + commitTS, err = i.forwardedTxnCommitTS(ctx, startTS) + if err != nil { + return 0, err + } + } + + for _, item := range metaMutations { + item.meta.CommitTS = commitTS + item.m.Value = kv.EncodeTxnMeta(item.meta) + } + if err := stampRequestMutationCommitTS(reqs, commitTS); err != nil { + return 0, err + } + return commitTS, nil +} + +func stampRequestMutationCommitTS(reqs []*pb.Request, commitTS uint64) error { + for _, r := range reqs { + if r == nil { + continue + } + if err := kv.StampMutationCommitTS(r.Mutations, commitTS); err != nil { + return errors.WithStack(err) + } + } + return nil +} + +func collectForwardedTxnMetas(reqs []*pb.Request) ([]forwardedTxnMetaToUpdate, uint64, error) { + metaMutations := make([]forwardedTxnMetaToUpdate, 0, len(reqs)) + var commitTS uint64 prefix := []byte(kv.TxnMetaPrefix) for _, r := range reqs { m, ok := forwardedTxnMetaMutation(r, prefix) @@ -238,24 +276,15 @@ func (i *Internal) fillForwardedTxnCommitTS(ctx context.Context, reqs []*pb.Requ continue } if meta.CommitTS != 0 { + if commitTS != 0 && commitTS != meta.CommitTS { + return nil, 0, errors.WithStack(kv.ErrInvalidRequest) + } + commitTS = meta.CommitTS continue } - metaMutations = append(metaMutations, metaToUpdate{m: m, meta: meta}) + metaMutations = append(metaMutations, forwardedTxnMetaToUpdate{m: m, meta: meta}) } - if len(metaMutations) == 0 { - return nil - } - - commitTS, err := i.forwardedTxnCommitTS(ctx, startTS) - if err != nil { - return err - } - - for _, item := range metaMutations { - item.meta.CommitTS = commitTS - item.m.Value = kv.EncodeTxnMeta(item.meta) - } - return nil + return metaMutations, commitTS, nil } // forwardedTxnCommitTS allocates a commit timestamp for a forwarded diff --git a/adapter/internal_test.go b/adapter/internal_test.go index 49802d20a..fcabc5356 100644 --- a/adapter/internal_test.go +++ b/adapter/internal_test.go @@ -2,6 +2,7 @@ package adapter import ( "context" + "encoding/binary" "testing" "github.com/bootjp/elastickv/kv" @@ -28,7 +29,7 @@ func TestStampTxnTimestamps_RejectsMaxStartTS(t *testing.T) { }, } - err := i.stampTxnTimestamps(context.Background(), reqs) + _, err := i.stampTxnTimestamps(context.Background(), reqs) require.ErrorIs(t, err, ErrTxnTimestampOverflow) } @@ -50,7 +51,7 @@ func TestFillForwardedTxnCommitTS_RejectsOverflow(t *testing.T) { }, } - err := i.fillForwardedTxnCommitTS(context.Background(), reqs, ^uint64(0)) + _, err := i.fillForwardedTxnCommitTS(context.Background(), reqs, ^uint64(0)) require.ErrorIs(t, err, ErrTxnTimestampOverflow) } @@ -73,11 +74,13 @@ func TestFillForwardedTxnCommitTS_AssignsCommitTS(t *testing.T) { }, } - require.NoError(t, i.fillForwardedTxnCommitTS(context.Background(), reqs, startTS)) + commitTS, err := i.fillForwardedTxnCommitTS(context.Background(), reqs, startTS) + require.NoError(t, err) meta, err := kv.DecodeTxnMeta(reqs[0].Mutations[0].Value) require.NoError(t, err) require.Equal(t, startTS+1, meta.CommitTS) + require.Equal(t, meta.CommitTS, commitTS) } func TestFillForwardedTxnCommitTS_PreservesExistingCommitTS(t *testing.T) { @@ -98,10 +101,12 @@ func TestFillForwardedTxnCommitTS_PreservesExistingCommitTS(t *testing.T) { }, } - require.NoError(t, i.fillForwardedTxnCommitTS(context.Background(), reqs, 10)) + commitTS, err := i.fillForwardedTxnCommitTS(context.Background(), reqs, 10) + require.NoError(t, err) meta, err := kv.DecodeTxnMeta(reqs[0].Mutations[0].Value) require.NoError(t, err) require.Equal(t, uint64(42), meta.CommitTS) + require.Equal(t, meta.CommitTS, commitTS) } func TestFillForwardedTxnCommitTS_AssignsCommitTSForOnePhaseTxn(t *testing.T) { @@ -128,11 +133,75 @@ func TestFillForwardedTxnCommitTS_AssignsCommitTSForOnePhaseTxn(t *testing.T) { }, } - require.NoError(t, i.fillForwardedTxnCommitTS(context.Background(), reqs, startTS)) + commitTS, err := i.fillForwardedTxnCommitTS(context.Background(), reqs, startTS) + require.NoError(t, err) meta, err := kv.DecodeTxnMeta(reqs[0].Mutations[0].Value) require.NoError(t, err) require.Equal(t, startTS+1, meta.CommitTS) + require.Equal(t, meta.CommitTS, commitTS) +} + +func TestFillForwardedTxnCommitTS_StampsCommitTSValueOffset(t *testing.T) { + t.Parallel() + + i := &Internal{} + startTS := uint64(10) + value := make([]byte, 16) + reqs := []*pb.Request{ + { + IsTxn: true, + Phase: pb.Phase_NONE, + Mutations: []*pb.Mutation{ + { + Op: pb.Op_PUT, + Key: []byte(kv.TxnMetaPrefix), + Value: kv.EncodeTxnMeta(kv.TxnMeta{PrimaryKey: []byte("k"), CommitTS: 0}), + }, + { + Op: pb.Op_PUT, + Key: []byte("k"), + Value: value, + CommitTsValueOffset: 4, + }, + }, + }, + } + + commitTS, err := i.fillForwardedTxnCommitTS(context.Background(), reqs, startTS) + require.NoError(t, err) + require.Equal(t, startTS+1, commitTS) + require.Equal(t, commitTS, binary.BigEndian.Uint64(value[4:12])) + require.Zero(t, reqs[0].Mutations[1].CommitTsValueOffset) +} + +func TestFillForwardedTxnCommitTS_PrepareAllowsAlreadyStampedOffsets(t *testing.T) { + t.Parallel() + + i := &Internal{} + value := make([]byte, 16) + mut := &pb.Mutation{ + Op: pb.Op_PUT, + Key: []byte("k"), + Value: value, + CommitTsValueOffset: 4, + } + require.NoError(t, kv.StampMutationCommitTS([]*pb.Mutation{mut}, 42)) + + reqs := []*pb.Request{ + { + IsTxn: true, + Phase: pb.Phase_PREPARE, + Ts: 10, + Mutations: []*pb.Mutation{mut}, + }, + } + + commitTS, err := i.fillForwardedTxnCommitTS(context.Background(), reqs, 10) + require.NoError(t, err) + require.Zero(t, commitTS) + require.Equal(t, uint64(42), binary.BigEndian.Uint64(value[4:12])) + require.Zero(t, mut.CommitTsValueOffset) } func TestStampTxnTimestamps_UsesSingleTxnStartTS(t *testing.T) { @@ -162,11 +231,13 @@ func TestStampTxnTimestamps_UsesSingleTxnStartTS(t *testing.T) { } reqs := []*pb.Request{prepare, commit} - require.NoError(t, i.stampTxnTimestamps(context.Background(), reqs)) + commitTS, err := i.stampTxnTimestamps(context.Background(), reqs) + require.NoError(t, err) require.Equal(t, uint64(9), prepare.Ts) require.Equal(t, uint64(9), commit.Ts) meta, err := kv.DecodeTxnMeta(commit.Mutations[0].Value) require.NoError(t, err) require.Greater(t, meta.CommitTS, uint64(9)) + require.Equal(t, meta.CommitTS, commitTS) } diff --git a/distribution/catalog.go b/distribution/catalog.go index 496340cf5..b92f85f23 100644 --- a/distribution/catalog.go +++ b/distribution/catalog.go @@ -22,7 +22,8 @@ const ( catalogVersionCodecVersion byte = 1 catalogRouteCodecVersionMin byte = 1 - catalogRouteCodecVersion byte = 1 + catalogRouteCodecVersionV2 byte = 2 + catalogRouteCodecVersion byte = catalogRouteCodecVersionV2 catalogScanPageSize = 256 catalogSaveMetaMutationCount = 2 @@ -76,6 +77,7 @@ type RouteDescriptor struct { GroupID uint64 State RouteState ParentRouteID uint64 + SplitAtHLC uint64 } // CatalogSnapshot is a point-in-time snapshot of the route catalog. @@ -183,12 +185,14 @@ func EncodeRouteDescriptor(route RouteDescriptor) ([]byte, error) { if route.End == nil { out = append(out, 0) + out = appendU64(out, route.SplitAtHLC) return out, nil } out = append(out, 1) out = appendU64(out, uint64(len(route.End))) out = append(out, route.End...) + out = appendU64(out, route.SplitAtHLC) return out, nil } @@ -212,7 +216,13 @@ func DecodeRouteDescriptor(raw []byte) (RouteDescriptor, error) { if err != nil { return RouteDescriptor{}, err } - if version == catalogRouteCodecVersion && r.Len() != 0 { + if version >= catalogRouteCodecVersionV2 { + route.SplitAtHLC, err = decodeRouteDescriptorSplitAtHLC(r) + if err != nil { + return RouteDescriptor{}, err + } + } + if version <= catalogRouteCodecVersion && r.Len() != 0 { return RouteDescriptor{}, errors.WithStack(ErrCatalogInvalidRouteRecord) } if err := validateRouteDescriptor(route); err != nil { @@ -224,21 +234,29 @@ func DecodeRouteDescriptor(raw []byte) (RouteDescriptor, error) { // Snapshot reads a consistent route catalog snapshot at the store's latest // known commit timestamp. func (s *CatalogStore) Snapshot(ctx context.Context) (CatalogSnapshot, error) { + if err := ensureCatalogStore(s); err != nil { + return CatalogSnapshot{}, err + } + return s.SnapshotAt(ctx, s.store.LastCommitTS()) +} + +// SnapshotAt reads a consistent route catalog snapshot at a specific MVCC +// timestamp. +func (s *CatalogStore) SnapshotAt(ctx context.Context, ts uint64) (CatalogSnapshot, error) { if err := ensureCatalogStore(s); err != nil { return CatalogSnapshot{}, err } ctx = contextOrBackground(ctx) - readTS := s.store.LastCommitTS() - version, err := s.versionAt(ctx, readTS) + version, err := s.versionAt(ctx, ts) if err != nil { return CatalogSnapshot{}, err } - routes, err := s.routesAt(ctx, readTS) + routes, err := s.routesAt(ctx, ts) if err != nil { return CatalogSnapshot{}, err } - return CatalogSnapshot{Version: version, Routes: routes, ReadTS: readTS}, nil + return CatalogSnapshot{Version: version, Routes: routes, ReadTS: ts}, nil } // Version reads only the durable catalog version at the latest commit @@ -375,6 +393,7 @@ func CloneRouteDescriptor(route RouteDescriptor) RouteDescriptor { GroupID: route.GroupID, State: route.State, ParentRouteID: route.ParentRouteID, + SplitAtHLC: route.SplitAtHLC, } } @@ -698,7 +717,8 @@ func routeDescriptorEqual(left, right RouteDescriptor) bool { bytes.Equal(left.End, right.End) && left.GroupID == right.GroupID && left.State == right.State && - left.ParentRouteID == right.ParentRouteID + left.ParentRouteID == right.ParentRouteID && + left.SplitAtHLC == right.SplitAtHLC } func appendU64(dst []byte, v uint64) []byte { @@ -712,6 +732,7 @@ func routeDescriptorEncodedSize(route RouteDescriptor) int { if route.End != nil { size += catalogUint64Bytes + len(route.End) } + size += catalogUint64Bytes return size } @@ -770,6 +791,17 @@ func decodeRouteDescriptorEnd(r *bytes.Reader) ([]byte, error) { return readU64LenBytes(r, endLen) } +func decodeRouteDescriptorSplitAtHLC(r *bytes.Reader) (uint64, error) { + if r.Len() < catalogUint64Bytes { + return 0, errors.WithStack(ErrCatalogInvalidRouteRecord) + } + var buf [catalogUint64Bytes]byte + if _, err := r.Read(buf[:]); err != nil { + return 0, errors.WithStack(err) + } + return binary.BigEndian.Uint64(buf[:]), nil +} + func appendRoutePage(out []*store.KVPair, page []*store.KVPair, prefix []byte) ([]*store.KVPair, []byte, bool) { var lastKey []byte for _, kvp := range page { diff --git a/distribution/catalog_test.go b/distribution/catalog_test.go index 96166b449..a3ffe852f 100644 --- a/distribution/catalog_test.go +++ b/distribution/catalog_test.go @@ -58,6 +58,7 @@ func TestRouteDescriptorCodecRoundTrip(t *testing.T) { GroupID: 3, State: RouteStateWriteFenced, ParentRouteID: 2, + SplitAtHLC: 123456789, } raw, err := EncodeRouteDescriptor(route) if err != nil { @@ -78,6 +79,7 @@ func TestRouteDescriptorCodecRoundTripNilEnd(t *testing.T) { GroupID: 2, State: RouteStateActive, ParentRouteID: 0, + SplitAtHLC: 987654321, } raw, err := EncodeRouteDescriptor(route) if err != nil { @@ -90,6 +92,42 @@ func TestRouteDescriptorCodecRoundTripNilEnd(t *testing.T) { assertRouteEqual(t, route, got) } +func TestRouteDescriptorCodecDecodesV1WithoutSplitAtHLC(t *testing.T) { + route := RouteDescriptor{ + RouteID: 5, + Start: []byte("a"), + End: []byte("m"), + GroupID: 1, + State: RouteStateActive, + ParentRouteID: 0, + } + raw := encodeRouteDescriptorV1ForTest(t, route) + + got, err := DecodeRouteDescriptor(raw) + if err != nil { + t.Fatalf("decode v1 route: %v", err) + } + assertRouteEqual(t, route, got) +} + +func TestRouteDescriptorCodecRejectsV1TrailingBytes(t *testing.T) { + route := RouteDescriptor{ + RouteID: 6, + Start: []byte("a"), + End: []byte("m"), + GroupID: 1, + State: RouteStateActive, + ParentRouteID: 0, + } + raw := encodeRouteDescriptorV1ForTest(t, route) + raw = append(raw, 0xff) + + _, err := DecodeRouteDescriptor(raw) + if !errors.Is(err, ErrCatalogInvalidRouteRecord) { + t.Fatalf("expected ErrCatalogInvalidRouteRecord, got %v", err) + } +} + func TestRouteDescriptorCodecRejectsTrailingBytes(t *testing.T) { route := RouteDescriptor{ RouteID: 1, @@ -111,6 +149,28 @@ func TestRouteDescriptorCodecRejectsTrailingBytes(t *testing.T) { } } +func TestRouteDescriptorCodecRejectsTruncatedSplitAtHLC(t *testing.T) { + route := RouteDescriptor{ + RouteID: 1, + Start: []byte("a"), + End: nil, + GroupID: 1, + State: RouteStateActive, + ParentRouteID: 0, + SplitAtHLC: 42, + } + raw, err := EncodeRouteDescriptor(route) + if err != nil { + t.Fatalf("encode route: %v", err) + } + raw = raw[:len(raw)-1] + + _, err = DecodeRouteDescriptor(raw) + if !errors.Is(err, ErrCatalogInvalidRouteRecord) { + t.Fatalf("expected ErrCatalogInvalidRouteRecord, got %v", err) + } +} + func TestRouteDescriptorCodecAcceptsForwardVersionTail(t *testing.T) { route := RouteDescriptor{ RouteID: 1, @@ -178,6 +238,27 @@ func TestRouteDescriptorCodecRejectsBelowMinimumVersion(t *testing.T) { } } +func TestRouteDescriptorCloneAndEqualIncludeSplitAtHLC(t *testing.T) { + route := RouteDescriptor{ + RouteID: 3, + Start: []byte("a"), + End: []byte("m"), + GroupID: 2, + State: RouteStateActive, + ParentRouteID: 1, + SplitAtHLC: 99, + } + clone := CloneRouteDescriptor(route) + assertRouteEqual(t, route, clone) + if !routeDescriptorEqual(route, clone) { + t.Fatal("expected clone to compare equal") + } + clone.SplitAtHLC++ + if routeDescriptorEqual(route, clone) { + t.Fatal("expected SplitAtHLC difference to compare unequal") + } +} + func TestCatalogRouteKeyHelpers(t *testing.T) { key := CatalogRouteKey(11) if !IsCatalogRouteKey(key) { @@ -642,6 +723,9 @@ func assertRouteEqual(t *testing.T, want, got RouteDescriptor) { if want.ParentRouteID != got.ParentRouteID { t.Fatalf("parent route id mismatch: want %d, got %d", want.ParentRouteID, got.ParentRouteID) } + if want.SplitAtHLC != got.SplitAtHLC { + t.Fatalf("split at HLC mismatch: want %d, got %d", want.SplitAtHLC, got.SplitAtHLC) + } if want.State != got.State { t.Fatalf("state mismatch: want %d, got %d", want.State, got.State) } @@ -652,3 +736,17 @@ func assertRouteEqual(t *testing.T, want, got RouteDescriptor) { t.Fatalf("end mismatch: want %q, got %q", want.End, got.End) } } + +func encodeRouteDescriptorV1ForTest(t *testing.T, route RouteDescriptor) []byte { + t.Helper() + route.SplitAtHLC = 0 + raw, err := EncodeRouteDescriptor(route) + if err != nil { + t.Fatalf("encode route: %v", err) + } + if len(raw) < catalogUint64Bytes+1 { + t.Fatalf("encoded route too short: %d", len(raw)) + } + raw[0] = catalogRouteCodecVersionMin + return raw[:len(raw)-catalogUint64Bytes] +} diff --git a/kv/commit_ts_patch.go b/kv/commit_ts_patch.go new file mode 100644 index 000000000..48bc02aa8 --- /dev/null +++ b/kv/commit_ts_patch.go @@ -0,0 +1,78 @@ +package kv + +import ( + "encoding/binary" + + pb "github.com/bootjp/elastickv/proto" + "github.com/cockroachdb/errors" +) + +const commitTSPatchLen = 8 + +// StampMutationCommitTS patches every mutation that embeds the resolved +// transaction commit timestamp in its value. Offset zero disables patching; +// non-zero offsets are byte offsets into the mutation value. +func StampMutationCommitTS(muts []*pb.Mutation, commitTS uint64) error { + for _, mut := range muts { + if mut == nil || mut.CommitTsValueOffset == 0 { + continue + } + if commitTS == 0 { + return errors.WithStack(ErrTxnCommitTSRequired) + } + if mut.Op != pb.Op_PUT { + return errors.WithStack(ErrInvalidRequest) + } + offset := mut.CommitTsValueOffset + if offset > uint64(len(mut.Value)) || uint64(len(mut.Value))-offset < commitTSPatchLen { + return errors.WithStack(ErrInvalidRequest) + } + binary.BigEndian.PutUint64(mut.Value[offset:offset+commitTSPatchLen], commitTS) + mut.CommitTsValueOffset = 0 + } + return nil +} + +func ValidateElemCommitTSPatches(elems []*Elem[OP], commitTS uint64) error { + if commitTS == 0 { + return rejectZeroCommitTSPatches(elems) + } + for _, elem := range elems { + if err := validateElemCommitTSPatch(elem); err != nil { + return err + } + } + return nil +} + +func rejectZeroCommitTSPatches(elems []*Elem[OP]) error { + for _, elem := range elems { + if elem != nil && elem.CommitTSValueOffset != 0 { + return errors.WithStack(ErrTxnCommitTSRequired) + } + } + return nil +} + +func validateElemCommitTSPatch(elem *Elem[OP]) error { + if elem == nil || elem.CommitTSValueOffset == 0 { + return nil + } + if elem.Op != Put { + return errors.WithStack(ErrInvalidRequest) + } + offset := elem.CommitTSValueOffset + if offset > uint64(len(elem.Value)) || uint64(len(elem.Value))-offset < commitTSPatchLen { + return errors.WithStack(ErrInvalidRequest) + } + return nil +} + +func StampGroupedMutationCommitTS(grouped map[uint64][]*pb.Mutation, commitTS uint64) error { + for _, muts := range grouped { + if err := StampMutationCommitTS(muts, commitTS); err != nil { + return err + } + } + return nil +} diff --git a/kv/commit_ts_patch_test.go b/kv/commit_ts_patch_test.go new file mode 100644 index 000000000..c46988c2d --- /dev/null +++ b/kv/commit_ts_patch_test.go @@ -0,0 +1,58 @@ +package kv + +import ( + "encoding/binary" + "testing" + + pb "github.com/bootjp/elastickv/proto" + "github.com/stretchr/testify/require" +) + +func TestStampMutationCommitTS(t *testing.T) { + t.Parallel() + + value := make([]byte, 16) + muts := []*pb.Mutation{{ + Op: pb.Op_PUT, + Value: value, + CommitTsValueOffset: 4, + }} + + require.NoError(t, StampMutationCommitTS(muts, 42)) + require.Equal(t, uint64(42), binary.BigEndian.Uint64(value[4:12])) + require.Zero(t, muts[0].CommitTsValueOffset) +} + +func TestStampMutationCommitTSRejectsInvalidPatch(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + mutation *pb.Mutation + commit uint64 + wantErr error + }{ + "zero commit": { + mutation: &pb.Mutation{Op: pb.Op_PUT, Value: make([]byte, 8), CommitTsValueOffset: 1}, + wantErr: ErrTxnCommitTSRequired, + }, + "non put": { + mutation: &pb.Mutation{Op: pb.Op_DEL, Value: make([]byte, 8), CommitTsValueOffset: 1}, + commit: 42, + wantErr: ErrInvalidRequest, + }, + "out of range": { + mutation: &pb.Mutation{Op: pb.Op_PUT, Value: make([]byte, 8), CommitTsValueOffset: 2}, + commit: 42, + wantErr: ErrInvalidRequest, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + + err := StampMutationCommitTS([]*pb.Mutation{tc.mutation}, tc.commit) + require.ErrorIs(t, err, tc.wantErr) + }) + } +} diff --git a/kv/coordinator.go b/kv/coordinator.go index f1b8a5627..5ee5c4f46 100644 --- a/kv/coordinator.go +++ b/kv/coordinator.go @@ -191,6 +191,7 @@ func marshalHLCLeaseRenew(ceilingMs int64) []byte { type CoordinateResponse struct { CommitIndex uint64 + CommitTS uint64 } type Coordinate struct { @@ -1078,6 +1079,9 @@ func (c *Coordinate) dispatchTxn(ctx context.Context, reqs []*Elem[OP], readKeys if commitTS <= startTS { return nil, errors.WithStack(ErrTxnCommitTSRequired) } + if err := ValidateElemCommitTSPatches(reqs, commitTS); err != nil { + return nil, err + } // ReadKeys are included in the Raft log entry so the FSM validates // read-write conflicts atomically under applyMu, eliminating the TOCTOU @@ -1096,6 +1100,7 @@ func (c *Coordinate) dispatchTxn(ctx context.Context, reqs []*Elem[OP], readKeys return &CoordinateResponse{ CommitIndex: r.CommitIndex, + CommitTS: commitTS, }, nil } @@ -1122,6 +1127,7 @@ func (c *Coordinate) dispatchRaw(ctx context.Context, req []*Elem[OP]) (*Coordin } return &CoordinateResponse{ CommitIndex: r.CommitIndex, + CommitTS: ts, }, nil } @@ -1231,6 +1237,7 @@ func (c *Coordinate) redirect(ctx context.Context, reqs *OperationGroup[OP]) (*C return &CoordinateResponse{ CommitIndex: r.CommitIndex, + CommitTS: r.CommitTs, }, nil } @@ -1283,10 +1290,15 @@ func (c *Coordinate) toForwardRequest(reqs []*pb.Request) *pb.ForwardRequest { func elemToMutation(req *Elem[OP]) *pb.Mutation { switch req.Op { case Put: + value := req.Value + if req.CommitTSValueOffset != 0 { + value = bytes.Clone(req.Value) + } return &pb.Mutation{ - Op: pb.Op_PUT, - Key: req.Key, - Value: req.Value, + Op: pb.Op_PUT, + Key: req.Key, + Value: value, + CommitTsValueOffset: req.CommitTSValueOffset, } case Del: return &pb.Mutation{ @@ -1326,6 +1338,12 @@ func onePhaseTxnRequestWithPrevCommit(startTS, commitTS, prevCommitTS uint64, pr for _, req := range reqs { muts = append(muts, elemToMutation(req)) } + if commitTS != 0 { + // stamp errors are validated before commit by dispatchTxn and + // Internal.Forward; this best-effort call keeps the helper nil-error + // compatible for existing tests that construct raw requests directly. + _ = StampMutationCommitTS(muts, commitTS) + } return &pb.Request{ IsTxn: true, Phase: pb.Phase_NONE, diff --git a/kv/coordinator_txn_test.go b/kv/coordinator_txn_test.go index c311cf818..05ab126d9 100644 --- a/kv/coordinator_txn_test.go +++ b/kv/coordinator_txn_test.go @@ -2,6 +2,7 @@ package kv import ( "context" + "encoding/binary" "testing" pb "github.com/bootjp/elastickv/proto" @@ -118,6 +119,31 @@ func TestCoordinateDispatchTxn_UsesProvidedCommitTS(t *testing.T) { require.Equal(t, commitTS, meta.CommitTS) } +func TestCoordinateDispatchTxn_StampsCommitTSValueOffsetWithoutMutatingElem(t *testing.T) { + t.Parallel() + + tx := &stubTransactional{} + c := &Coordinate{ + transactionManager: tx, + clock: NewHLC(), + } + + startTS := uint64(10) + commitTS := uint64(25) + value := make([]byte, 16) + resp, err := c.dispatchTxn(context.Background(), []*Elem[OP]{ + {Op: Put, Key: []byte("k"), Value: value, CommitTSValueOffset: 4}, + }, nil, startTS, commitTS, 0, 0) + require.NoError(t, err) + require.Equal(t, commitTS, resp.CommitTS) + require.Len(t, tx.reqs, 1) + + got := tx.reqs[0][0].Mutations[1].Value + require.Equal(t, commitTS, binary.BigEndian.Uint64(got[4:12])) + require.Zero(t, tx.reqs[0][0].Mutations[1].CommitTsValueOffset) + require.Zero(t, binary.BigEndian.Uint64(value[4:12])) +} + func TestCoordinateDispatchTxn_PassesReadKeysToRaftEntry(t *testing.T) { t.Parallel() diff --git a/kv/sharded_coordinator.go b/kv/sharded_coordinator.go index b8b67dba5..8b0cf2c83 100644 --- a/kv/sharded_coordinator.go +++ b/kv/sharded_coordinator.go @@ -1092,6 +1092,9 @@ func (c *ShardedCoordinator) dispatchTxn(ctx context.Context, startTS uint64, co if err != nil { return nil, err } + if err := ValidateElemCommitTSPatches(elems, commitTS); err != nil { + return nil, err + } if len(gids) == 1 && c.allReadKeysInShard(readKeys, gids[0]) { // Fast path: all mutations and read keys are in a single shard. @@ -1101,6 +1104,9 @@ func (c *ShardedCoordinator) dispatchTxn(ctx context.Context, startTS uint64, co // preserving SSI. return c.dispatchSingleShardTxn(ctx, startTS, commitTS, prevCommitTS, primaryKey, gids[0], elems, readKeys, observedRouteVersion) } + if err := StampGroupedMutationCommitTS(grouped, commitTS); err != nil { + return nil, err + } return c.dispatchMultiShardTxn(ctx, startTS, commitTS, prevCommitTS, primaryKey, grouped, gids, readKeys, observedRouteVersion) } @@ -1168,7 +1174,7 @@ func (c *ShardedCoordinator) dispatchMultiShardTxn(ctx context.Context, startTS, if err != nil { return nil, errors.WithStack(err) } - return &CoordinateResponse{CommitIndex: maxIndex}, nil + return &CoordinateResponse{CommitIndex: maxIndex, CommitTS: commitTS}, nil } func (c *ShardedCoordinator) resolveTxnCommitTS(ctx context.Context, startTS, commitTS uint64) (uint64, error) { @@ -1219,7 +1225,7 @@ func (c *ShardedCoordinator) dispatchSingleShardTxn(ctx context.Context, startTS if resp == nil { return &CoordinateResponse{}, nil } - return &CoordinateResponse{CommitIndex: resp.CommitIndex}, nil + return &CoordinateResponse{CommitIndex: resp.CommitIndex, CommitTS: commitTS}, nil } type preparedGroup struct { @@ -1980,6 +1986,9 @@ func (c *ShardedCoordinator) txnLogs(ctx context.Context, reqs *OperationGroup[O if err != nil { return nil, err } + if err := StampGroupedMutationCommitTS(grouped, commitTS); err != nil { + return nil, err + } return buildTxnLogs(reqs.StartTS, commitTS, grouped, gids, reqs.ObservedRouteVersion) } diff --git a/kv/sharded_coordinator_txn_test.go b/kv/sharded_coordinator_txn_test.go index 964f2713f..9fd229937 100644 --- a/kv/sharded_coordinator_txn_test.go +++ b/kv/sharded_coordinator_txn_test.go @@ -2,6 +2,7 @@ package kv import ( "context" + "encoding/binary" "errors" "sync" "testing" @@ -107,12 +108,14 @@ func TestShardedCoordinatorDispatchTxn_CrossShardPhasesAndCommitIndex(t *testing }, 1, NewHLC(), nil) startTS := uint64(10) + value1 := make([]byte, 16) + value2 := make([]byte, 16) resp, err := coord.Dispatch(context.Background(), &OperationGroup[OP]{ IsTxn: true, StartTS: startTS, Elems: []*Elem[OP]{ - {Op: Put, Key: []byte("b"), Value: []byte("v1")}, - {Op: Put, Key: []byte("x"), Value: []byte("v2")}, + {Op: Put, Key: []byte("b"), Value: value1, CommitTSValueOffset: 4}, + {Op: Put, Key: []byte("x"), Value: value2, CommitTSValueOffset: 4}, }, }) require.NoError(t, err) @@ -134,9 +137,9 @@ func TestShardedCoordinatorDispatchTxn_CrossShardPhasesAndCommitIndex(t *testing require.Len(t, g1Prepare.Mutations, 2) require.Len(t, g2Prepare.Mutations, 2) require.Equal(t, []byte("b"), g1Prepare.Mutations[1].Key) - require.Equal(t, []byte("v1"), g1Prepare.Mutations[1].Value) require.Equal(t, []byte("x"), g2Prepare.Mutations[1].Key) - require.Equal(t, []byte("v2"), g2Prepare.Mutations[1].Value) + require.Zero(t, g1Prepare.Mutations[1].CommitTsValueOffset) + require.Zero(t, g2Prepare.Mutations[1].CommitTsValueOffset) prepareMeta1 := requestTxnMeta(t, g1Prepare) prepareMeta2 := requestTxnMeta(t, g2Prepare) @@ -166,6 +169,12 @@ func TestShardedCoordinatorDispatchTxn_CrossShardPhasesAndCommitIndex(t *testing require.Zero(t, commitMeta2.LockTTLms) require.Greater(t, commitMeta1.CommitTS, startTS) require.Equal(t, commitMeta1.CommitTS, commitMeta2.CommitTS) + require.Equal(t, commitMeta1.CommitTS, binary.BigEndian.Uint64(g1Prepare.Mutations[1].Value[4:12])) + require.Equal(t, commitMeta1.CommitTS, binary.BigEndian.Uint64(g2Prepare.Mutations[1].Value[4:12])) + require.Zero(t, g1Commit.Mutations[1].CommitTsValueOffset) + require.Zero(t, g2Commit.Mutations[1].CommitTsValueOffset) + require.Zero(t, binary.BigEndian.Uint64(value1[4:12])) + require.Zero(t, binary.BigEndian.Uint64(value2[4:12])) } func TestShardedCoordinatorDispatchTxn_SingleShardUsesOnePhase(t *testing.T) { diff --git a/kv/transcoder.go b/kv/transcoder.go index afc52bed3..1bb4ac455 100644 --- a/kv/transcoder.go +++ b/kv/transcoder.go @@ -19,6 +19,10 @@ type Elem[T OP] struct { Op T Key []byte Value []byte + // CommitTSValueOffset, when non-zero, asks the coordinator or forwarded + // leader to stamp the resolved transaction commit timestamp into Value at + // this byte offset before committing the mutation. + CommitTSValueOffset uint64 } // OperationGroup is a group of operations that should be executed atomically. diff --git a/proto/distribution.pb.go b/proto/distribution.pb.go index c51d2aac2..b6847386e 100644 --- a/proto/distribution.pb.go +++ b/proto/distribution.pb.go @@ -444,6 +444,7 @@ type RouteDescriptor struct { RaftGroupId uint64 `protobuf:"varint,4,opt,name=raft_group_id,json=raftGroupId,proto3" json:"raft_group_id,omitempty"` State RouteState `protobuf:"varint,5,opt,name=state,proto3,enum=RouteState" json:"state,omitempty"` ParentRouteId uint64 `protobuf:"varint,6,opt,name=parent_route_id,json=parentRouteId,proto3" json:"parent_route_id,omitempty"` + SplitAtHlc uint64 `protobuf:"varint,7,opt,name=split_at_hlc,json=splitAtHlc,proto3" json:"split_at_hlc,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -520,6 +521,13 @@ func (x *RouteDescriptor) GetParentRouteId() uint64 { return 0 } +func (x *RouteDescriptor) GetSplitAtHlc() uint64 { + if x != nil { + return x.SplitAtHlc + } + return 0 +} + type SplitJobBracketProgress struct { state protoimpl.MessageState `protogen:"open.v1"` BracketId uint64 `protobuf:"varint,1,opt,name=bracket_id,json=bracketId,proto3" json:"bracket_id,omitempty"` @@ -1141,14 +1149,16 @@ const file_distribution_proto_rawDesc = "" + "\rraft_group_id\x18\x03 \x01(\x04R\vraftGroupId\"\x15\n" + "\x13GetTimestampRequest\"4\n" + "\x14GetTimestampResponse\x12\x1c\n" + - "\ttimestamp\x18\x01 \x01(\x04R\ttimestamp\"\xc3\x01\n" + + "\ttimestamp\x18\x01 \x01(\x04R\ttimestamp\"\xe5\x01\n" + "\x0fRouteDescriptor\x12\x19\n" + "\broute_id\x18\x01 \x01(\x04R\arouteId\x12\x14\n" + "\x05start\x18\x02 \x01(\fR\x05start\x12\x10\n" + "\x03end\x18\x03 \x01(\fR\x03end\x12\"\n" + "\rraft_group_id\x18\x04 \x01(\x04R\vraftGroupId\x12!\n" + "\x05state\x18\x05 \x01(\x0e2\v.RouteStateR\x05state\x12&\n" + - "\x0fparent_route_id\x18\x06 \x01(\x04R\rparentRouteId\"\xb0\x02\n" + + "\x0fparent_route_id\x18\x06 \x01(\x04R\rparentRouteId\x12 \n" + + "\fsplit_at_hlc\x18\a \x01(\x04R\n" + + "splitAtHlc\"\xb0\x02\n" + "\x17SplitJobBracketProgress\x12\x1d\n" + "\n" + "bracket_id\x18\x01 \x01(\x04R\tbracketId\x12\x16\n" + diff --git a/proto/distribution.proto b/proto/distribution.proto index 33cc9cb62..f2199d909 100644 --- a/proto/distribution.proto +++ b/proto/distribution.proto @@ -42,6 +42,7 @@ message RouteDescriptor { uint64 raft_group_id = 4; RouteState state = 5; uint64 parent_route_id = 6; + uint64 split_at_hlc = 7; } enum SplitJobPhase { diff --git a/proto/internal.pb.go b/proto/internal.pb.go index f5ec91269..720eee66a 100644 --- a/proto/internal.pb.go +++ b/proto/internal.pb.go @@ -129,12 +129,18 @@ func (Phase) EnumDescriptor() ([]byte, []int) { } type Mutation struct { - state protoimpl.MessageState `protogen:"open.v1"` - Op Op `protobuf:"varint,1,opt,name=op,proto3,enum=Op" json:"op,omitempty"` - Key []byte `protobuf:"bytes,2,opt,name=key,proto3" json:"key,omitempty"` - Value []byte `protobuf:"bytes,3,opt,name=value,proto3" json:"value,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + Op Op `protobuf:"varint,1,opt,name=op,proto3,enum=Op" json:"op,omitempty"` + Key []byte `protobuf:"bytes,2,opt,name=key,proto3" json:"key,omitempty"` + Value []byte `protobuf:"bytes,3,opt,name=value,proto3" json:"value,omitempty"` + // commit_ts_value_offset, when non-zero, asks the committing leader to write + // the resolved transaction commit timestamp into value[offset:offset+8] as a + // big-endian uint64 before the request is committed. This is used for values + // that durably embed their own creation timestamp but still need leader-side + // timestamp allocation after redirect/retry. + CommitTsValueOffset uint64 `protobuf:"varint,4,opt,name=commit_ts_value_offset,json=commitTsValueOffset,proto3" json:"commit_ts_value_offset,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *Mutation) Reset() { @@ -188,6 +194,13 @@ func (x *Mutation) GetValue() []byte { return nil } +func (x *Mutation) GetCommitTsValueOffset() uint64 { + if x != nil { + return x.CommitTsValueOffset + } + return 0 +} + type Request struct { state protoimpl.MessageState `protogen:"open.v1"` IsTxn bool `protobuf:"varint,1,opt,name=is_txn,json=isTxn,proto3" json:"is_txn,omitempty"` @@ -385,6 +398,7 @@ type ForwardResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Success bool `protobuf:"varint,1,opt,name=success,proto3" json:"success,omitempty"` CommitIndex uint64 `protobuf:"varint,2,opt,name=commit_index,json=commitIndex,proto3" json:"commit_index,omitempty"` + CommitTs uint64 `protobuf:"varint,3,opt,name=commit_ts,json=commitTs,proto3" json:"commit_ts,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -433,6 +447,13 @@ func (x *ForwardResponse) GetCommitIndex() uint64 { return 0 } +func (x *ForwardResponse) GetCommitTs() uint64 { + if x != nil { + return x.CommitTs + } + return 0 +} + type RelayPublishRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Channel []byte `protobuf:"bytes,1,opt,name=channel,proto3" json:"channel,omitempty"` @@ -533,11 +554,12 @@ var File_internal_proto protoreflect.FileDescriptor const file_internal_proto_rawDesc = "" + "\n" + - "\x0einternal.proto\"G\n" + + "\x0einternal.proto\"|\n" + "\bMutation\x12\x13\n" + "\x02op\x18\x01 \x01(\x0e2\x03.OpR\x02op\x12\x10\n" + "\x03key\x18\x02 \x01(\fR\x03key\x12\x14\n" + - "\x05value\x18\x03 \x01(\fR\x05value\"\xca\x01\n" + + "\x05value\x18\x03 \x01(\fR\x05value\x123\n" + + "\x16commit_ts_value_offset\x18\x04 \x01(\x04R\x13commitTsValueOffset\"\xca\x01\n" + "\aRequest\x12\x15\n" + "\x06is_txn\x18\x01 \x01(\bR\x05isTxn\x12\x1c\n" + "\x05phase\x18\x02 \x01(\x0e2\x06.PhaseR\x05phase\x12\x0e\n" + @@ -549,10 +571,11 @@ const file_internal_proto_rawDesc = "" + "\brequests\x18\x01 \x03(\v2\b.RequestR\brequests\"M\n" + "\x0eForwardRequest\x12\x15\n" + "\x06is_txn\x18\x01 \x01(\bR\x05isTxn\x12$\n" + - "\brequests\x18\x02 \x03(\v2\b.RequestR\brequests\"N\n" + + "\brequests\x18\x02 \x03(\v2\b.RequestR\brequests\"k\n" + "\x0fForwardResponse\x12\x18\n" + "\asuccess\x18\x01 \x01(\bR\asuccess\x12!\n" + - "\fcommit_index\x18\x02 \x01(\x04R\vcommitIndex\"I\n" + + "\fcommit_index\x18\x02 \x01(\x04R\vcommitIndex\x12\x1b\n" + + "\tcommit_ts\x18\x03 \x01(\x04R\bcommitTs\"I\n" + "\x13RelayPublishRequest\x12\x18\n" + "\achannel\x18\x01 \x01(\fR\achannel\x12\x18\n" + "\amessage\x18\x02 \x01(\fR\amessage\"8\n" + diff --git a/proto/internal.proto b/proto/internal.proto index 53215dc91..fb8f70e55 100644 --- a/proto/internal.proto +++ b/proto/internal.proto @@ -25,6 +25,12 @@ message Mutation { Op op = 1; bytes key = 2; bytes value = 3; + // commit_ts_value_offset, when non-zero, asks the committing leader to write + // the resolved transaction commit timestamp into value[offset:offset+8] as a + // big-endian uint64 before the request is committed. This is used for values + // that durably embed their own creation timestamp but still need leader-side + // timestamp allocation after redirect/retry. + uint64 commit_ts_value_offset = 4; } enum Phase { @@ -68,6 +74,7 @@ message ForwardRequest { message ForwardResponse { bool success = 1; uint64 commit_index = 2; + uint64 commit_ts = 3; } message RelayPublishRequest {