Skip to content

fix(planetscale): prevent silent data loss on vstream context canceled errors #76

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
dev-menon wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix(planetscale): prevent silent data loss on vstream context canceled errors #76

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7f46e6b
Handle Context cancel failures
dev-menon Mar 20, 2026
3b50609
Update connect_client.go
dev-menon Mar 20, 2026
42e703a
Address requested comments
dev-menon Mar 24, 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
44 changes: 38 additions & 6 deletions lib/connect_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,9 +152,13 @@ func (p connectClient) Read(ctx context.Context, logger DatabaseLogger, ps Plane
}
if err != nil {
if s, ok := status.FromError(err); ok {
// if the error is anything other than server timeout, keep going
// codes.DeadlineExceeded is a server-side timeout and is safe to retry
// from the last checkpointed cursor. codes.Canceled means the client's
// own context was canceled (e.g., Fivetran requesting shutdown) — the
// parent context is already done, so retrying would just burn through
// backoff sleeps without making progress. Treat it as non-retryable.
if s.Code() != codes.DeadlineExceeded {
logger.Warning(fmt.Sprintf("%vGot error [%v] with message [%q], Returning with cursor :[%v] after non-timeout error", preamble, s.Code(), err, currentPosition))
logger.Warning(fmt.Sprintf("%vGot error [%v] with message [%q], Returning with cursor :[%v] after non-retryable error", preamble, s.Code(), err, currentPosition))

// Check for binlog expiration error and reset cursor for historical sync
if IsBinlogsExpirationError(err) {
Expand Down Expand Up @@ -291,13 +295,34 @@ func (p connectClient) sync(ctx context.Context, logger DatabaseLogger, tableNam

// stop when we've reached the well known stop position for this sync session.
watchForVgGtidChange := false
// lastSafeTC tracks the last cursor position where rows were confirmed written to
// the destination. VStream delivers rows and their VGTID in the same SyncResponse,
// so tc and lastSafeTC only diverge on cursor-only responses (heartbeats, DDL events,
// or transactions for unrelated tables that carry no rows for this table). On any
// mid-stream error, returning lastSafeTC avoids checkpointing a cursor that has no
// corresponding rows in the destination, forcing a safe re-stream from the last
// confirmed write position.
lastSafeTC := tc
for {

res, err := c.Recv()
if err != nil {
return tc, err
if errors.Is(err, io.EOF) {
// Natural end of stream: all rows up to tc have been delivered.
return tc, err
Copy link
Copy Markdown
Contributor

@nickvanw nickvanw Mar 20, 2026

Choose a reason for hiding this comment

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

The EOF special-case here is correct — on clean stream termination, all rows have been delivered, so returning tc is right.

But for the non-EOF error path on the next line (return lastSafeTC, err): this is overly conservative given the actual protocol behavior. Since the edge-gateway sends cursor + rows in the same SyncResponse, tc and lastSafeTC will differ only when the most recent response was a cursor-only heartbeat/DDL/unrelated-table event — meaning there are no rows at that position to lose. Returning lastSafeTC just forces unnecessary re-streaming of events that have no rows for this table.

Not a correctness issue (re-streaming is safe), but worth understanding that this isn't preventing data loss — it's adding redundant replay.

}
// Mid-stream error (e.g. context canceled): return the last cursor
// where rows were confirmed written, not the potentially advanced tc.
return lastSafeTC, err
}

// Determine whether this response carries any row data before processing.
// Cursor-only responses (heartbeats, DDL events, unrelated-table transactions)
// carry no rows for this table. lastSafeTC is only advanced when rows have been
// delivered, so a mid-stream error after a cursor-only response re-streams from
// the last position where rows were confirmed written.
rowsInResponse := len(res.Result) > 0 || len(res.Deletes) > 0 || len(res.Updates) > 0

if onResult != nil {
for _, insertedRow := range res.Result {
qr := sqltypes.Proto3ToResult(insertedRow)
Expand All @@ -307,7 +332,7 @@ func (p connectClient) sync(ctx context.Context, logger DatabaseLogger, tableNam
}
sqlResult.Rows = append(sqlResult.Rows, row)
if err := onResult(sqlResult, OpType_Insert); err != nil {
return tc, status.Error(codes.Internal, "unable to serialize row")
return lastSafeTC, status.Error(codes.Internal, "unable to serialize row")
}
}
}
Expand All @@ -320,7 +345,7 @@ func (p connectClient) sync(ctx context.Context, logger DatabaseLogger, tableNam
}
Copy link
Copy Markdown
Contributor

@nickvanw nickvanw Mar 20, 2026

Choose a reason for hiding this comment

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

This is a genuine bug fix — the old code returned nil here, and Read() immediately dereferences the result:

currentPosition, err = p.sync(...)
if currentPosition.Position != "" {  // nil pointer panic

Same for the update serialization error at line 354. This fix should be extracted and merged on its own — it doesn't depend on the lastSafeTC or codes.Canceled changes.

sqlResult.Rows = append(sqlResult.Rows, row)
if err := onResult(sqlResult, OpType_Delete); err != nil {
return nil, status.Error(codes.Internal, "unable to serialize row")
return lastSafeTC, status.Error(codes.Internal, "unable to serialize row")
}
}
}
Expand All @@ -333,13 +358,20 @@ func (p connectClient) sync(ctx context.Context, logger DatabaseLogger, tableNam
After: serializeQueryResult(update.After),
}
if err := onUpdate(updatedRow); err != nil {
return nil, status.Error(codes.Internal, "unable to serialize update")
return lastSafeTC, status.Error(codes.Internal, "unable to serialize update")
}
}
}

if res.Cursor != nil {
tc = res.Cursor
// Only advance lastSafeTC when rows have been written at this cursor.
// Cursor-only responses (heartbeats, DDL, unrelated-table events) leave
// lastSafeTC unchanged so a mid-stream error replays from the last
// confirmed write rather than an intermediate cursor with no row data.
if rowsInResponse {
lastSafeTC = tc
}
}

// Because of the ordering of events in a vstream
Expand Down