Skip to content

distribution: add split HLC route codec #1066

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bootjp wants to merge 6 commits into
base: main
Choose a base branch
from
Open

distribution: add split HLC route codec #1066

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4e3bff7
distribution: add split HLC route codec
bootjp Jul 11, 2026
9856a5c
distribution: avoid reflective split HLC decode
bootjp Jul 11, 2026
14962c3
distribution: stamp split child routes
bootjp Jul 11, 2026
eea7ea0
distribution: keep split timestamps leader-assigned
bootjp Jul 11, 2026
d0a7c4c
distribution: return exact split snapshot
bootjp Jul 13, 2026
90d9410
Merge remote-tracking branch 'origin/main' into design/hotspot-split-…
bootjp Jul 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 59 additions & 20 deletions adapter/distribution_server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())
}
Expand All @@ -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 {
Expand All @@ -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)

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Reload the exact split version

Because saveSplitResultViaCoordinator reloads any snapshot with Version >= nextVersion, a concurrent SplitRange on another node can advance the catalog again before this line runs. If that follow-up split targets one of these just-created children, the newer snapshot no longer contains left.RouteID or right.RouteID, so splitChildrenFromSnapshot returns an Internal error even though this split already committed successfully; load the snapshot at this transaction's commit timestamp/exact catalog version before extracting the response descriptors.

Useful? React with 👍 / 👎.

if err != nil {
return nil, err
}

return &pb.SplitRangeResponse{
CatalogVersion: saved.Version,
Left: toProtoRouteDescriptor(left),
Right: toProtoRouteDescriptor(right),
Left: toProtoRouteDescriptor(savedLeft),

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Include SplitAtHLC in route descriptors

When SplitRange is called through the Distribution API, savedLeft now contains the leader-stamped SplitAtHLC, but toProtoRouteDescriptor/pb.RouteDescriptor still omit that field, so the response serializes the same descriptor shape as before. Any caller that relies on the SplitRange/ListRoutes RPC output to seed or reconstruct split cooldowns will see no durable split timestamp and can treat just-created children as not cooling down; add a split_at_hlc field to distribution.proto and populate it here.

Useful? React with 👍 / 👎.

Right: toProtoRouteDescriptor(savedRight),
}, nil
}

Expand Down Expand Up @@ -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(
Expand All @@ -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]{
Expand All @@ -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())
Expand All @@ -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 {
Expand All @@ -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
Expand All @@ -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,
)
}

Expand Down Expand Up @@ -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{
Expand All @@ -390,6 +426,7 @@ func splitCatalogRoutes(
GroupID: parent.GroupID,
State: parent.State,
ParentRouteID: parent.RouteID,
SplitAtHLC: splitAtHLC,
}
right := distribution.RouteDescriptor{
RouteID: rightID,
Expand All @@ -398,6 +435,7 @@ func splitCatalogRoutes(
GroupID: parent.GroupID,
State: parent.State,
ParentRouteID: parent.RouteID,
SplitAtHLC: splitAtHLC,
}
return left, right
}
Expand Down Expand Up @@ -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 {
Expand All @@ -453,6 +491,7 @@ func toProtoRouteDescriptor(route distribution.RouteDescriptor) *pb.RouteDescrip
RaftGroupId: route.GroupID,
State: toProtoRouteState(route.State),
ParentRouteId: route.ParentRouteID,
SplitAtHlc: route.SplitAtHLC,
}
}

Expand Down
Loading
Loading