feat(multidb): pipeline, tx pipeline and autopipeline support - #3951
feat(multidb): pipeline, tx pipeline and autopipeline support#3951ndyakov wants to merge 20 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR extends MultiDBClient to fully satisfy the UniversalClient surface and integrates pipelining and autopipelining into the MultiDB core, so batched operations resolve the active database at execution time and can fail over/retry as needed.
Changes:
- Add
hooksMixinsupport toMultiDBClientand wire MultiDB-level hooks into command, pipeline, and tx-pipeline processing. - Implement MultiDB-aware
Pipeline/TxPipelineprocessing and cachedAutoPipeline/AsyncAutoPipelineinstances with lifecycle management onClose. - Add/extend tests to cover pipeline routing, failover retry behavior, per-command outcome recording, tx at-most-once behavior, and autopipeline caching/closure.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
multidb.go |
Embeds hooksMixin, adds cached autopipeliner fields, routes Process through hooks, and drains autopipeliners in Close. |
multidb_test.go |
Extends the test hook to count pipeline batches and simulate pipeline-wide failures. |
multidb_pipeline.go |
Introduces MultiDB pipeline/tx-pipeline execution, Do, Watch, SSubscribe, PoolStats, and AutoPipeline APIs. |
multidb_autopipeline_test.go |
Adds coverage for pipeline and autopipeline routing, failover/retry, at-most-once tx behavior, and instance caching/closure. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
f926af2 to
34eeef4
Compare
|
@codex review |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 34eeef4fc0
ℹ️ About Codex in GitHub
Codex has been enabled to automatically 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 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
34eeef4 to
5360697
Compare
|
@codex review |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.
Suppressed comments (1)
multidb_pipeline.go:174
- AutoPipeline/AsyncAutoPipeline refuse cluster member databases only at creation time. If a MultiDB client creates an AutoPipeliner and later a cluster member becomes active (e.g., added via AddDatabase or selected by failover), the existing autopipeliner will still dispatch batches to that cluster member, bypassing the intended refusal and the cluster-specific autopipeline safeguards mentioned in the comment above.
Consider adding an execution-time guard in the unhooked autopipeliner entry points (process/processPipeline) to fail fast when the active database is a cluster client, so an already-created autopipeliner can’t silently run in an unsupported mode after topology changes.
// process / processPipeline are the unhooked base entry points required by
// the autopipeliner's backend interface (cmdableClient); the hook-wrapped
// variants come from hooksMixin.
func (c *MultiDBClient) process(ctx context.Context, cmd Cmder) error {
return c.core.process(ctx, cmd)
}
// hookCount reports one more hook than are installed on the MultiDBClient
// itself: member clients can carry their own hooks (AddDatabaseHook), which
// the autopipeliner cannot see, and its async dispatcher only arms the
// batch-executor self-deadlock guards when hookCount is non-zero. Always
// reporting at least one keeps those guards armed for member-level hooks
// that read command results.
func (c *MultiDBClient) hookCount() int {
return c.hooksMixin.hookCount() + 1
}
func (c *MultiDBClient) processPipeline(ctx context.Context, cmds []Cmder) error {
return c.core.processPipeline(ctx, cmds)
}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 53606975bc
ℹ️ About Codex in GitHub
Codex has been enabled to automatically 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 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
5360697 to
0a47d9c
Compare
|
@codex review |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.
Suppressed comments (3)
multidb_pipeline.go:103
- processPipeline uses the name transportFailures for the value returned by recordBatchOutcomes, but that count includes any outcomeFailure (including retryable server replies), not only transport errors. Consider renaming the local variable so the retry decision is clearly based on availability failures, matching classifyOutcome.
err := db.processPipelineHook(ctx, cmds)
transportFailures := c.recordBatchOutcomes(db, cmds)
if transportFailures == 0 {
// Only server replies (or clean success) — done, whatever the
multidb_pipeline.go:39
- The counter variable in recordBatchOutcomes is named transportFailures, but it is incremented for any outcomeFailure (including retryable server replies via shouldRetry), not only transport-level errors. Renaming it to reflect "availability failures" will make the intent match classifyOutcome and avoid misleading future changes.
func (c *multidbCore) recordBatchOutcomes(db *multidbDatabase, cmds []Cmder) int {
transportFailures := 0
for _, cmd := range cmds {
err := cmd.rawErr()
switch classifyOutcome(err) {
multidb_pipeline.go:29
- The comment says "error reply from the server ... counts as a success" and that the function returns the number of "transport-level failures", but the implementation uses classifyOutcome(err) and counts outcomeFailure, which also includes retryable server replies (shouldRetry), not just transport errors. Please update the comment to reflect that retryable server replies (e.g. LOADING/READONLY/CLUSTERDOWN) are treated as availability failures and included in the returned count.
This issue also appears in the following locations of the same file:
- line 35
- line 99
// mirroring the single-command path: an error reply from the server proves
// the database is reachable and counts as a success; transport-level errors
// count as failures; client-side errors (context cancellation, deterministic
// local rejections per shouldRetry) record nothing and do not trigger
// failover. It returns the number of transport-level failures.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 0a47d9caa6
ℹ️ About Codex in GitHub
Codex has been enabled to automatically 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 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
aa40a1c to
00af763
Compare
0a47d9c to
6ea2e43
Compare
|
@codex review |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.
Suppressed comments (1)
multidb_pipeline.go:30
- The docstring says this function returns the number of "transport-level failures", but the implementation increments the counter for every outcomeFailure (which also includes retryable server replies per shouldRetry). This mismatch is easy to misread when reasoning about retries/failover semantics; please align the comment with classifyOutcome.
// recordBatchOutcomes records one breaker/detector outcome per command,
// mirroring the single-command path: an error reply from the server proves
// the database is reachable and counts as a success; transport-level errors
// count as failures; client-side errors (context cancellation, deterministic
// local rejections per shouldRetry) record nothing and do not trigger
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 6ea2e43f7c
ℹ️ About Codex in GitHub
Codex has been enabled to automatically 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 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
00af763 to
0221df0
Compare
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 73efb7721f
ℹ️ About Codex in GitHub
Codex has been enabled to automatically 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 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 07cabd006e
ℹ️ About Codex in GitHub
Codex has been enabled to automatically 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 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
|
@codex review |
|
Codex Review: Didn't find any major issues. 👍 Reviewed commit: ℹ️ About Codex in GitHubCodex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback". |
MultiDBClient now satisfies UniversalClient and the autopipeliner backend interface: batches resolve the active database at exec time, feed the circuit breaker and failure detector per command, and are retried against the newly selected database after a failover. MULTI/EXEC pipelines are executed at most once and never retried automatically.
Batches containing a non-retryable command (RawWriteToCmd and friends) execute at most once, and the Watch documentation now states that a transaction stays bound to the member that was active at call time.
- batch outcome recording reads rawErr (never Err) so the async autopipeline dispatcher cannot await the batch it is completing - client-side errors in batches record nothing and disable the retry, matching the single-command path - tx pipeline outcome recording excludes the synthetic MULTI/EXEC envelope commands - AutoPipeline/AsyncAutoPipeline refuse cluster member databases (the cluster-specific autopipeline wiring would be bypassed) - Watch attempts a failover first when the active member is unhealthy, and its docs spell out the member binding and hook behavior - hookCount reports one extra hook so the async batch-executor guards stay armed for member-level hooks
- batch outcome recording uses the shared classifier: retryable server replies count as failures, ErrCrossSlot and other local errors are neutral - availability errors from exhausted failover attempts overwrite stale per-command transport errors (setCmdsErr fills only empty slots) - AddDatabase refuses cluster members while an autopipeliner instance exists, closing the gap around the creation-time cluster check
- a closed autopipeliner no longer blocks AddDatabase(cluster) forever (the cached slot is checked for liveness, matching getOrCreate) - the cluster-member check moved inside the pipeliner build function and AddDatabase holds autopipelinerMu across check+add, so creation and cluster additions are fully serialized
- neutral-only batches release the half-open probe slot reserved by the breaker gate - pipeline and tx-pipeline paths reject work with ErrClosed after Close, matching the single-command path - WATCH gates on the non-reserving circuit read (it never records or releases half-open slots) and returns ErrClosed after Close - the AddDatabase/AutoPipeliner.Close drain race window is documented
Suppress the member client's own pipeline retry loop for MultiDB transactions by marking the synthetic MULTI as NoRetry (EXEC may have committed before a transport error; at-most-once must hold for the whole stack), re-enter the breaker admission gate after a batch/tx failover so a half-open member's bounded probe slots are respected, and return context errors before the failover gate in TxPipeline and Watch so doomed operations cannot advance failover state.
Mark every command of a MultiDB transaction NoRetry (cluster members trim the MULTI/EXEC envelope before their retry check, so a marker on the synthetic MULTI alone is lost there), check the failure detector before the breaker admission gate in the batch and tx paths so a detector-routed failover cannot leak a half-open probe slot, overwrite stale transport errors with the context error when a retried batch is canceled between attempts, and classify the batch-level error for unstamped commands so a hook-aborted batch records nothing instead of phantom successes.
Fall back to the batch-level error only when NO command was stamped (an executed batch always stamps at least the failing command), fixing the round-eight substitution that could turn a successful prefix into phantom transport failures and replay it; stamp that error onto the unstamped commands so per-command inspection and exhausted-retry returns see it; and mark batch successes as recovery traffic for the failover escalation chain, matching the single-command path.
Record batch failures before successes so a half-open member's failed recovery batch cannot close the circuit off its own successful prefix, and replace the stamped-commands inference with an execution marker (flipped by the standalone and cluster execution paths the moment a batch may reach the wire): a post-exec hook error on an executed batch no longer fabricates transport failures and replays applied writes, while pre-exec hook aborts still surface on every command.
Treat hook-served batches (nil without execution) as neutral — the caller keeps its results, but nothing reached Redis, so no breaker or detector signal is recorded and the gate's probe slot is released; stamp the batch error onto every command an aborting hook left unstamped so partially-stamped aborts cannot fabricate successes; and bound consecutive gate rejections in the batch retry loop, mirroring the single-command path, so all-members-half-open-and-full surfaces ErrTemporarilyNotAvailable instead of busy-looping failovers.
Reject the HIMPORT command family in the batch paths too (pipeline builders dispatch through their own cmdable and bypass the direct overrides), skip success recording entirely for unexecuted batches so hook-fabricated replies cannot close a half-open circuit, and apply failure-first outcome ordering only to half-open recovery probes — in the closed state the arrival order stands, or a stale sub-threshold failure could combine with a batch failure and open a healthy member.
Reject only the HIMPORT commands themselves in plain pipelines (an autopipeliner flush coalesces unrelated callers, and poisoning the whole batch failed innocent commands), keep whole-batch rejection for transactions (atomic by definition), and turn the TxPipeline gate into the same bounded failover loop the other paths use — a denied probe slot now retries another member instead of rejecting the transaction, which is safe because no MULTI/EXEC has been sent while gating.
Classify batch outcomes with the per-command blocking-timeout rule (cmd.readTimeout) instead of a blanket retryTimeout=true, surface the HIMPORT rejection as the Exec error when the kept commands all succeed, and make errMultiDBHImport a RedisError so the autopipeliner's sequential dispatch treats it as a per-command verdict rather than a fatal abort that poisons the sub-batches queued behind it.
SSubscribe rejects a Close race through the MultiDB PubSub path (which fails dials with terminal ErrClosed) and registers the requested shard channels on the no-active fallback like Subscribe/PSubscribe do, and a mixed pipeline with a rejected HIMPORT reports the positionally first failed command as its Exec error.
Mirrors the single-command retry loop: a Close landing mid-retry reports ErrClosed instead of escalating through the drained membership.
The batch guards match HIMPORT by command name too (raw Do/NewCmd submissions build plain *Cmd values the marker interface misses), and the Watch gate reserves a half-open probe slot via IsAllowed — released after the call, since WATCH outcomes never record — so MaxHalfOpenRequests bounds concurrent transactions against a recovering member.
A half-open candidate with its probe budget exhausted failed Watch with ErrTemporarilyNotAvailable after a single failover attempt while a healthy member was still selectable. The gate now loops like the pipeline and tx-pipeline gates (bounded by the rejection cap); no WATCH is sent while the gate is still choosing, so at-most-once holds.
A Watch admitted while the breaker was closed reserved no probe slot, but its deferred release ran anyway — a transaction outliving a later open -> half-open transition freed a slot a real recovery probe held, letting MaxHalfOpenRequests be exceeded. The gate now uses Allow and releases only reserved admissions. Early pipeline exits (Close, context, availability) returned the stamped error even when a rejected HIMPORT preceded the stamped commands positionally; every exit now reports the positionally first error over the original slice, like the executed path.
Batch gates admitted closed-state work via IsAllowed, then settled through unconditional ReleaseHalfOpen (hook-served and all-neutral batches) and RecordSuccess — a batch outliving a later open -> half-open transition freed probe slots it never reserved. Both gates use Allow now and recordBatchOutcomes settles by the reservation: unreserved successes count via RecordExternalSuccess, unreserved releases are skipped.
|
@codex review |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 8 out of 8 changed files in this pull request and generated no new comments.
Suppressed comments (1)
multidb_pipeline.go:32
- The doc comment for recordBatchOutcomes says any server error reply counts as success and that the return value is "transport-level failures", but the implementation uses classifyOutcome (which treats retryable server replies like LOADING/READONLY/CLUSTERDOWN as failures) and increments the counter for any outcomeFailure, not only transport errors. This mismatch can mislead future changes around retry/failover semantics.
// mirroring the single-command path: an error reply from the server proves
// the database is reachable and counts as a success; transport-level errors
// count as failures; client-side errors (context cancellation, deterministic
// local rejections per shouldRetry) record nothing and do not trigger
// failover. It returns the number of transport-level failures.
|
Codex Review: Didn't find any major issues. 🚀 Reviewed commit: ℹ️ About Codex in GitHubCodex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback". |
Stacked on the MultiDBClient orchestration PR.
Makes
MultiDBClienta fullUniversalClientand wires batching through theMultiDB core:
hooksMixinonMultiDBClient(MultiDB-levelAddHook), satisfying theautopipeliner's backend interface — one
AutoPipelinerinstance survivesfailover; every flushed batch resolves the active database at exec time
processPipeline: batch attempt loop bounded byCommandRetries;per-command outcome recording (server error replies count as healthy);
transport-failed batches are retried against the newly selected database
processTxPipeline: MULTI/EXEC executed at most once, never auto-retried(EXEC may have committed before the connection broke)
Pipeline/Pipelined,TxPipeline/TxPipelined,Do,Watch(active-bound, aborts on failover),
SSubscribe,PoolStatsAutoPipeline/AsyncAutoPipeline(+WithOptions) with the same cachedinstance semantics as
*Client;Closedrains pipeliners before memberclients shut down
Known trade-off: batches reaching a cluster member are split per node by the
member's own pipeline path (correct, not slot-shard-optimized).
Note
High Risk
Changes failover, circuit-breaker recording, and at-most-once semantics for pipelines and transactions across the MultiDB stack; incorrect batch retry or health recording could duplicate writes or mask outages.
Overview
Makes
MultiDBClienta drop-inUniversalClientby adding MultiDB-level hooks,Do,Pipeline/Pipelined,TxPipeline/TxPipelined,Watch,SSubscribe,PoolStats, and cachedAutoPipeline/AsyncAutoPipeline(withClosedraining pipeliners first).New
multidb_pipelinecore routes batches to the active member at exec time:processPipelineretries on transport failure (bounded byCommandRetries), records breaker/detector outcomes per command, and re-gates through half-open circuit admission;processTxPipelineandWatchgate similarly but execute MULTI/EXEC at most once. Tx paths mark every commandsetNoRetryso member-level retries cannot replay committed transactions.baseCmdgains anoRetryflag (cloned with commands) soNoRetry()is data-driven—used for streaming writes and MultiDB tx batches.pipelineExecutedKeyin context lets MultiDB distinguish hook-aborted batches from executed ones when recording health signals.HIMPORT is rejected on batch paths (per-command in plain pipelines, whole tx rejected); errors are
proto.RedisErrorso autopipeline coalescing does not abort unrelated commands. Cluster + autopipeline is refused at creation time;AddDatabase(cluster)is blocked while live autopipeliners exist.Extensive integration tests cover failover retries, probe-slot accounting, hook-served batches, and escalation errors.
Reviewed by Cursor Bugbot for commit 18947f9. Bugbot is set up for automated code reviews on this repo. Configure here.