vstreamclient: framework for robust + simple usage

[derekperkins]

Description

Implementation for #17221. I did a walkthrough in the March community meeting, starting at 7:18.

The actual code was 95% written by me, tests were 95% written by AI

Related Issue(s)

• Fixes RFC: VStream client reference implementation #17221

Checklist

• "Backport to:" labels have been added if this change should be back-ported to release branches
• If this change is to be back-ported to previous releases, a justification is included in the PR description
• Tests were added or are not required
• Did the new or modified tests pass consistently locally and on CI?
• Documentation was added or is not required

Deployment Notes

Adds the public go/vt/vstreamclient package for checkpointed Vitess event consumption.

• Checkpoints require a separate unsharded state keyspace. Construction requires readable VSchema sharding metadata and executes CREATE TABLE IF NOT EXISTS, even for a pre-provisioned table.
• All source primaries, eligible stream tablets, and writer/replication sessions must use FULL row images throughout the retained replay range. New probes each shard's primary and configured stream tablet type; unverifiable targets fail closed unless explicitly bypassed after out-of-band verification.
• Run claims ownership after receiving its first successful VStream response. Initial receive failures and startup timeouts leave the incumbent's owner token untouched. State-row fencing protects checkpoints; concurrent consumers still require serialized handoffs or sink-enforced fencing/version checks.
• Explicit starting positions must be concrete and cover every configured source shard for the selected tablet type. Symbolic positions and TablePKs copy cursors are rejected.
• The default liveness window is now ten seconds. Liveness failures cancel immediately; WithHeartbeatSeconds overrides WithFlags.HeartbeatInterval regardless of option order, including when the supplied flags leave the interval at zero.
• Graceful shutdown can complete at a ROLLBACK boundary without waiting for a later heartbeat.
• Construction rejects heartbeat intervals and timeout multipliers whose combined liveness window exceeds the maximum supported duration.

[vitess-bot]

Review Checklist

Hello reviewers! 👋 Please follow this checklist when reviewing this Pull Request.

General

• Ensure that the Pull Request has a descriptive title.
• Ensure there is a link to an issue (except for internal cleanup and flaky test fixes), new features should have an RFC that documents use cases and test cases.

Tests

• Bug fixes should have at least one unit or end-to-end test, enhancement and new features should have a sufficient number of tests.

Documentation

• Apply the release notes (needs details) label if users need to know about this change.
• New features should be documented.
• There should be some code comments as to why things are implemented the way they are.
• There should be a comment at the top of each new or modified test to explain what the test does.

New flags

• Is this flag really necessary?
• Flag names must be clear and intuitive, use dashes (-), and have a clear help text.

If a workflow is added or modified:

• Each item in Jobs should be named in order to mark it as required.
• If the workflow needs to be marked as required, the maintainer team must be notified.

Backward compatibility

• Protobuf changes should be wire-compatible.
• Changes to _vt tables and RPCs need to be backward compatible.
• RPC changes should be compatible with vitess-operator
• If a flag is removed, then it should also be removed from vitess-operator and arewefastyet, if used there.
• vtctl command output order should be stable and awk-able.

[rohit-nayak-ps]

Overall it looks very good. I have mentioned some issues below. Once we address those, and whatever you feel are needed as part of this, we should merge and address the several TODOs noted in the future.

I am still undecided about where to locate it. I will let others weigh in here.

I had some issues running the unit tests:

• I had to drop the state table before running TestVStreamClient for the second time. 
• I could only run one of the tests at a time. Looks like the first test doesn't clean up the state and hence the second fails, if I just run go test. Also it will be good to name TestVStreamClient to TestVStreamClientBasic or some such to make it easier to run each test separately
• I think it might be easier not to require the state table by default (or provide some mechanism to clean it up either by default or by setting a flag separately). We could store the state locally for the duration of the test in-memory. Maybe if we fix the previous issues this may not be necessary though ...
• It is difficult to debug this in an IDE because of different short timeouts. While trying to figure out some local failures various contexts kept expiring even after setting a few of the values to high numbers. Not sure what the best solution for this is: maybe move all timeout constants to a common place so anyone developing/debugging can modify it there. Maybe this is just needed for the initial phase, but others who fork this might also run into it while testing their changes.

Also:

• We can just send "" as the vgtid to start with the copy phase. No need to generate the per-shard vgtid done in New()
• We should also get the failing tests working. Maybe it just needs a rebase. If not, let us know if you need help with any eventual failing tests.
• Not sure if you had a look at the e2e tests in go/test/endtoend/vreplication. Those allow creating a cluster if you wanted to create an e2e test for this.
• Can you explain the specific tests for which you wanted to use non-exported functions from vtgate/vttablet. go/test/endtoend/vreplication has tests where entire clusters are created. Do those help?

[derekperkins]

@rohit-nayak-ps thanks for the review!

Yes, the unit tests weren't meant to be true unit tests, just a way to test functionality, with me truncating and restarting regularly. Whatever the new test harness turns out to be will be easily testable. I just wanted to make sure I had an API that people thought was good and that I didn't need to do any major refactoring before spinning it up.

We can just send "" as the vgtid to start with the copy phase. No need to generate the per-shard vgtid done in New()

I'm not sure how to do that. The VGtid struct only has []*ShardGtid. It didn't accept me sending through an empty VGtid (maybe I only tested nil?), so this seemed like the only way.

go/test/endtoend/vreplication seems like it has what I need for testing at first glance. I hope to get the tests set up and PR passing this weekend.

[rohit-nayak-ps]

Yes, the unit tests weren't meant to be true unit tests, just a way to test functionality, with me truncating and restarting regularly. Whatever the new test harness turns out to be will be easily testable. I just wanted to make sure I had an API that people thought was good and that I didn't need to do any major refactoring before spinning it up.

Understood. I see you are already working on this.

We can just send "" as the vgtid to start with the copy phase. No need to generate the per-shard vgtid done in New()
I'm not sure how to do that. The VGtid struct only has []*ShardGtid. It didn't accept me sending through an empty VGtid (maybe I only tested nil?), so this seemed like the only way.

See example in end to end test TestVStreamCurrent:

var shardGtids []*binlogdatapb.ShardGtid
	vgtid := &binlogdatapb.VGtid{}
	shardGtids = append(shardGtids, &binlogdatapb.ShardGtid{
		Keyspace: "ks",
		Shard:    "-80",
		Gtid:     "current",
	})
	shardGtids = append(shardGtids, &binlogdatapb.ShardGtid{
		Keyspace: "ks",
		Shard:    "80-",
		Gtid:     "current",
	})
	vgtid.ShardGtids = shardGtids
	filter := &binlogdatapb.Filter{
		Rules: []*binlogdatapb.Rule{{
			Match:  "t1",
			Filter: "select * from t1",
		}},
	}
	flags := &vtgatepb.VStreamFlags{}
	reader, err := gconn.VStream(ctx, topodatapb.TabletType_PRIMARY, vgtid, filter, flags)

[derekperkins]

See example in end to end test TestVStreamCurrent:

That vstream test code seems to be initializing each shard. I was patterning my solution on that and other test code I saw, except I am handling generic keyspaces / shards. Am I missing something?

https://github.com/vitessio/vitess/blob/ecb37200d31c389883129b5d141c279ecdfcd688/go/vt/vstreamclient/state.go#L100-L106

[github-actions]

This PR is being marked as stale because it has been open for 30 days with no activity. To rectify, you may do any of the following:

• Push additional commits to the associated branch.
• Remove the stale label.
• Add a comment indicating why it is not stale.

If no action is taken within 7 days, this PR will be closed.

[github-actions]

This PR was closed because it has been stale for 7 days with no activity.

[Copilot]

Pull request overview

Copilot reviewed 23 out of 23 changed files in this pull request and generated no new comments.

[devin-ai-integration]

Note

This report is out of date. Scroll down for Devin Review's latest report on this PR.

Devin Review found 1 new potential issue.

[devin-ai-integration]

🔍 Type declarations remain ungrouped

The new files declare types separately, despite the repository rule requiring one grouped type declaration at each file’s top.

---

Was this helpful? React with 👍 or 👎 to provide feedback.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 0704e1be49

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

Include the stream name in the skipped-check warning

When a process constructs multiple VStream clients, this warning does not identify which stream disabled the row-image safety check, leaving operators unable to trace a potential corruption risk back to the responsible workflow. Add the configured stream name as a structured field, as the other liveness warnings in this package already do.

AGENTS.md reference: AGENTS.md:L116-L117

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Mark the first response before claiming ownership

When the first Recv succeeds near the startup deadline but the state-row claim is slow, the watchdog still sees lastEventProcessedAtUnixNano == 0 and isProcessingEvents == false because that guard is not set until handleEvents, after this ownership call. It can therefore cancel a stream that already received a response; if the claim then completes successfully, this run exits while leaving its owner token persisted, fencing the healthy incumbent without an active replacement. Mark the response or ownership phase as active before issuing the state write.

AGENTS.md reference: AGENTS.md:L119-L124

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Flush old-schema rows before invoking DDL hooks

When a DDL hook applies the schema change to a downstream sink, it runs here before the DDL branch calls flush. Rows from earlier committed transactions may still be buffered because they did not meet the duration or row threshold, so the client then sends old-schema rows to a sink whose schema has already changed, which can reject or misinterpret them. Flush the pre-DDL batch before invoking a hook intended to apply the DDL, or provide a post-flush DDL callback.

AGENTS.md reference: AGENTS.md:L119-L124

Useful? React with 👍 / 👎.

[Copilot]

Pull request overview

Copilot reviewed 24 out of 24 changed files in this pull request and generated 1 comment.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: c904050432

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

Document that MaxRowsPerFlush is not a hard buffer cap

This promises that at most MaxRowsPerFlush rows remain buffered, but handleRowEvent appends an entire row event and flushing occurs only at safe transaction boundaries. With TransactionChunkSize enabled, a transaction can span many responses and grow currentBatch far beyond this value before COMMIT; a single large response can exceed it as well. Operators relying on this statement for memory sizing can therefore still encounter unbounded transaction-sized memory use, so the documentation should describe the setting as a flush trigger and output chunk limit rather than a maximum buffer size.

Useful? React with 👍 / 👎.

[devin-ai-integration]

Devin Review found 1 new potential issue.

[devin-ai-integration]

🔴 Partial updates mislabel column values

When partial updates omit leading columns, MakeRowTrusted assigns remaining values the wrong column types. Custom scanners can reject or corrupt valid updates.

Prompt for agents

In go/vt/vstreamclient/table.go, TableConfig.handleRowEvent converts compact partial after-images with sqltypes.MakeRowTrusted(shard.fields, rc.After). That helper pairs row lengths with fields positionally, but RowChange.DataColumns identifies which full-schema columns are actually present. If an omitted column precedes a present one, the resulting sqltypes.Value gets the omitted column's type. Build the row for VStreamScanner implementations using DataColumns to select the corresponding field types while preserving the compact value order. Full images and delete before-images must retain their current behavior. Add a scanner test where a partial after-image omits the first field and includes a later field of a different SQL type.

---

Was this helpful? React with 👍 or 👎 to provide feedback.