mirror of
https://github.com/redis/go-redis.git
synced 2026-10-24 07:59:20 +03:00
c994cfc85f
* ci: run PR workflows for feature/** base branches PRs stacked against feature/autopipeline/integration (and any feature/** integration branch) didn't trigger build, e2e, govulncheck or codeql because their pull_request.branches filters only matched master/v9/ndyakov/**/etc. A pull_request event filter matches the BASE branch, so PRs targeting a feature branch were skipped. Add 'feature/**' to those filters. * feat(autopipeline): #1 base types and client wiring (#3864) * perf(proto): read payload+CRLF in one shot in ReadStringInto When the caller-supplied buffer has room for the payload plus the trailing CRLF, read both in a single io.ReadFull instead of a separate line read. This is the zero-copy fast path used by GetToBuffer. * feat(autopipeline): add AutoPipeliner with deferred typed results Introduce AutoPipeliner, a background batcher that coalesces concurrent commands from many goroutines into Redis pipelines for high throughput. Engine (autopipeline.go): - Sharded command queue (per-shard mutex + flusher + FIFO semaphore), round-robin enqueue to cut contention under concurrency. - Batch completion signalled by a single close(batch.done) instead of per-command channel sends. - AutoPipelineConfig{MaxBatchSize, MaxConcurrentBatches, Unordered, MaxFlushDelay, AdaptiveDelay}. Default is ordered (MaxConcurrentBatches=1). Validate() rejects MaxConcurrentBatches>1 without Unordered:true so callers can't silently lose ordering; NewAutoPipeliner panics on an invalid config. Deferred results (command.go): - baseCmd gains a ready channel + setReady/await/rawErr. Every exported value-accessor (Val/Result/Text/Int/Bytes/Bool/Float/Time/Scan/First/Item/ ...) awaits the result before reading cmd.val/cmd.err; delegating accessors inherit the await through the method they call. Execution-path reads use rawErr() to avoid awaiting the very batch they produce (self-deadlock). - Cmder interface extended with setReady/await/rawErr. Pooling (pipeline.go) reuses Pipeline/cmd-slice objects across batches. osscluster.go and redis.go use rawErr() on the pipeline execution path. * feat(autopipeline): return an error from constructors instead of panicking newAutoPipeliner / NewAutoPipeliner validated the config and panicked on a bad one. Since these run from the post-init AutoPipeline / AsyncAutoPipeline calls (not client construction), a panic is the wrong failure mode — a deterministic config mistake should surface as a returned error the caller can handle. Return (*AutoPipeliner, error); the config is rejected before any flusher goroutine starts, so the error path leaks nothing. * fix(autopipeline): don't cap the shutdown flush with a hardcoded 2s timeout flushBatchSliceShutdown wrapped the final drain in context.WithTimeout(2s). Because the effective wire deadline is min(ctx deadline, conn timeout), that 2s capped everything — including a relaxed read/write timeout set by maintnotifications during a failover/migration. In-flight commands on shutdown would then time out at 2s instead of honoring the relaxed window the relaxation was designed to protect. ap.ctx is already cancelled at this point, so use a plain context.Background() with no artificial deadline, matching the normal flush path and a plain client Exec: the connection's ReadTimeout/WriteTimeout (and any relaxed override) governs the wire deadline. A user who wants shutdown bounded sets those, as for any command. * fix(autopipeline): correct permit-wait timeout and error in flushBatchSlice The flush waited a hardcoded 5s for a concurrency permit and, on any non-nil result, errored every command in the batch with ErrClosed. Two problems: - ErrClosed is wrong when the wait simply timed out — the client is not closed. Callers saw a misleading closed error for a deadline. - 5s is below a maintnotifications relaxed window, so a legitimately slow in-flight batch (e.g. during a failover) holding the permits could make waiters fail spuriously. Distinguish a real shutdown (ap.ctx cancelled -> ErrClosed) from the backstop firing (-> the timeout error as-is), and raise the backstop to 30s — well above the default ReadTimeout and a relaxed window. It remains only a safety net: every permit holder releases via defer and each batch Exec is bounded by the conn timeout, and the wait still ends immediately when Close cancels ap.ctx. * docs(autopipeline): fix usage example to the error-returning API The AutoPipeliner type doc showed 'ap := client.AutoPipeline()', the old single-return form. AutoPipeline / AsyncAutoPipeline now return (*AutoPipeliner, error); update the example to assign and check err so the base reflects the final API surface. * refactor(autopipeline): drop orphan AutoPipeline() accessor stubs (*Pipeline).AutoPipeline() returned a silent nil and (*AutoPipeliner). AutoPipeline() returned itself, with a comment claiming they satisfy the Cmdable interface. They don't — neither Cmdable nor Pipeliner declares AutoPipeline(), so these methods satisfy nothing and the nil-returning Pipeline stub is a latent nil-deref footgun. The real entry points are Client/ClusterClient.AutoPipeline(config...) (*AutoPipeliner, error). Remove both stubs; the var _ Cmdable assertion still holds. * fix(autopipeline): Process honors mode; unexport constructor; doc fixes - Process() now routes through ap.cmdable like Do and the typed methods, so on a blocking autopipeliner it blocks instead of always running async. Previously it hardcoded processAsync, diverging from every other entry point. - Unexport NewAutoPipeliner -> newAutoPipeliner. The exported form took an unexported cmdableClient param and skipped cluster slot-sharding, silently degrading a *ClusterClient to cross-node batches. The public entry points (Client/ClusterClient.AutoPipeline/AsyncAutoPipeline) wire sharding correctly. - Fix stale docs: Do no longer claims it always 'returns immediately' (it blocks in blocking mode); TxPipeline/TxPipelined drop the false 'will panic for other clients' note (cmdableClient embeds Cmdable, so TxPipeline always exists). * ci: run PR workflows for feature/** base branches PRs stacked against a feature/** integration branch (e.g. feature/autopipeline/integration) didn't trigger build, e2e, govulncheck or codeql: a pull_request event filter matches the BASE branch, and those filters only listed master/v9/ndyakov/** etc. Add 'feature/**' so stacked PRs run CI. * fix(json): avoid autopipeline deadlock and stale reads in JSON commands JSONCmd/JSONSliceCmd readReply called baseCmd.Err(), which now awaits the autopipeline batch's done channel. readReply runs inside that batch's Exec, before done is closed, so a deferred JSON command would deadlock. Use rawErr() there instead. Also add the missing await() to JSONCmd.Expanded and JSONSliceCmd.Val/Result so deferred callers block for the result instead of reading empty/stale data. Reported by Cursor Bugbot on PR #3864. * ci: exempt autopipeline engine files from the unused linter The autopipeline stack defines its engine on lower branches and wires the callers in on top, so those symbols read as unused until the stack is assembled, failing golangci-lint on the base/pool PRs. Scope the unused exemption to autopipeline.go and pipeline.go only; unused stays enforced everywhere else. * docs(autopipeline): MaxBatchSize default is 200, not 100 DefaultAutoPipelineConfig and the zero-value fill both use 200; the field doc said 100. Reported by Copilot on PR #3865. * fix(search): await deferred FTHybrid cursor/raw accessors; clarify MaxBatchSize - FTHybridCmd.CursorResult/CursorVal/RawVal/RawResult read their fields without await(), unlike Val/Result. On a deferred autopipeline that returns empty/stale data before the command executes. Add await() to match. (readReply sets the fields directly, so this can't re-introduce the batch-Exec deadlock.) - Clarify the MaxBatchSize doc: it is a soft accumulation threshold, not a hard cap; under burst the batch can exceed it and execute as one larger (safe) pipeline. Reported by Cursor Bugbot on PR #3864. * fix(autopipeline): drain shards in Close after flushers exit A command could pass enqueue's under-lock closed-recheck moments before Close's CompareAndSwap, then append to a shard AFTER that shard's flusher had already drained and exited — leaving its batch.done unclosed, so the caller's blocking accessor hung forever (lost command). Close now drains every shard once more after wg.Wait() (flushers provably gone). The drain loop checks queue emptiness UNDER s.mu instead of via the unlocked s.Len(), so it cannot miss a command appended concurrently. This pairs with the closed-recheck in enqueue: s.mu serializes them, so either the late enqueue appends first and the sweep flushes it, or the sweep empties first and the enqueue then sees closed==true and rejects with ErrClosed. Fixes the hang in TestAutoPipelineCloseRace. * address pr comments * search needs to wait on raw result * feat(autopipeline): #2 add pipeline pool (#3865) * perf(autopipeline): flush once the queue stops growing on the default window accumulateBatch waited the full defaultAccumulateWindow (200µs) whenever the queue was below MaxBatchSize, even for a single blocking caller with nothing left to coalesce, so every low-concurrency command paid the whole window. The doc described a "stops growing" early exit that was never implemented. Implement it for the implicit default window only: a burst timer fires when no new command arrives for defaultAccumulateGap (20µs); each enqueue resets it, so concurrent load keeps coalescing up to the window cap while a lone caller flushes ~one gap after enqueuing. An explicit MaxFlushDelay / AdaptiveDelay is an intentional accumulation window and is still waited in full. defaultAccumulateWindow becomes a var so tests can enlarge it deterministically. * feat(autopipeline): default the blocking face to a single ordered batch stream DefaultBlockingAutoPipelineConfig used MaxConcurrentBatches:50 "for throughput", but that is pessimal. With many flush permits, each command finds a free permit and flushes on its own before others accumulate, collapsing batch size toward one command per round-trip — i.e. non-pipelined throughput at high latency. Backpressure from a single in-flight batch (MaxConcurrentBatches:1) instead lets callers whose commands return re-enqueue and flush together as the next batch, so batches stay deep (a near-continuous, double-buffered pipeline). Measured (get=80/set=20, 64B, localhost, 128 blocking workers): before (b=50): ~59k ops/s, p50 2.9ms, avg batch 1.6 after (b=1): ~420k ops/s, p50 0.30ms, avg batch ~94 Higher throughput AND ~10x lower latency, and it is naturally ordered. The async default (DefaultAutoPipelineConfig) is already MaxConcurrentBatches:1; for maximum throughput use AsyncAutoPipeline with a window of in-flight commands (inflight>1). * feat(pool): add optional dedicated pipeline connection pool Add an opt-in separate connection pool for pipeline/autopipeline execution (PipelinePoolSize), so heavy pipelining doesn't starve the regular command pool. withPipelineConn draws from it when configured and falls back to the main pool otherwise, draining push notifications before returning a connection. Attach the existing per-pool listeners to the dedicated pool as well, so its connections get the same lifecycle handling as the main pool: - streaming credentials re-auth hook - maintnotifications handoff hook (MOVING/MIGRATING). Because that hook's failed-handoff removal target is its own single pool field, the pipeline pool gets an independent hook (Manager.InitPoolHookForPool) bound to it, rather than sharing the main pool's hook — otherwise a pipeline-pool connection that failed handoff would be removed from the wrong pool. The extra hook shares the Manager, so MOVING/MIGRATING tracking stays central, and is shut down and detached on Manager.Close. * docs(options): correct pipeline buffer recommendation to 64-128 KiB The recommended 512 KiB overstated the benefit. Benchmarks show pipeline throughput climbs from the 32 KiB default to ~64 KiB and then plateaus; beyond ~128 KiB there is no further gain and very large buffers (>=512 KiB) can regress throughput and waste memory. The only effect of the buffer is holding one batch's wire bytes so it flushes in a single syscall, so size it to roughly MaxBatchSize x average-command-bytes. Update the recommendation, example, and memory-impact figures accordingly. * feat(pool): register the dedicated pipeline pool with OTel The optional pipeline connection pool was created, used and closed but never registered with the OTel recorder, so its connection-count / pending-request metrics were invisible. Add a pipelinePool param to otel.RegisterPools / UnregisterPools (registered as a regular pool under a _pipeline name suffix) and pass c.pipelinePool from NewClient and the client teardown. nil when no pipeline pool is configured, so default clients are unaffected. * fix(pool): limiter on pipeline-pool path, hook-attach race, test addr Address Copilot review on PR #3865: - withPipelineConn now honors opt.Limiter (Allow() before acquiring, ReportResult() on every exit incl. the early init/re-acquire failures), mirroring getConn/releaseConn. Enabling the pipeline pool no longer silently bypasses throttling and failure reporting. - InitPoolHookForPool now holds hooksMu across the closed-check, append and AddPoolHook, closing a (construction-only) window where Close could tear the hook down between append and attach, leaking it. AddPoolHook is lock-free and never re-enters the manager, so this is deadlock-safe. - pipeline_buffer_test.go uses the shared redisAddr instead of a hardcoded localhost:6379 so it respects the env/CI Redis address. * perf(autopipeline): decouple shard count from MaxConcurrentBatches Shard count was derived as min(GOMAXPROCS, MaxConcurrentBatches, 16), which silently fragmented the command queue as soon as a user raised MaxConcurrentBatches for throughput: 16 shards each accumulate a sliver of the load, so batches collapse toward one command per flush and pipelining disappears. Measured on a 14-core host (get=80/set=20, 64B, 128 goroutines, async window 8): MaxConcurrentBatches=50 meant 16 shards and 413k ops/s at 1.4ms p50, while the identical permit budget on ONE shard reaches 1.75M ops/s at 0.51ms p50 — the fragmentation, not the permit count, was the regression. Add AutoPipelineConfig.NumShards (0 = auto: one shard) and stop deriving the shard count from the permit budget. One deep queue plus 2-4 permits is the sweet spot: batch N+1 accumulates and executes while batch N's replies are in flight (async, window 8: 1.78M ops/s at 0.53ms p50 versus 1.0M at 1.0ms with a single permit). Every shard keeps at least one permit, so the effective global concurrency is max(NumShards, MaxConcurrentBatches). numAutoPipelineShards stays as the cluster-wiring default, where slot-routed shards keep each batch on one node. * perf(autopipeline): stripe the enqueue queue where ordering permits Profiling the windowed-GET throughput benchmark showed the shard enqueue mutex as the hottest lock in the engine: with 128 concurrent callers on one shard, lock/unlock slow paths (parking and waking contended waiters) accounted for roughly half of all CPU samples. Split the shard's queue into 8 stripes, each with its own mutex, queue, and batch-completion signal; enqueue picks a stripe round-robin, and the flusher drains every stripe into one merged pipeline so batches stay as deep as before. Striping applies where reordering is impossible or waived: unordered configs, and every blocking-face shard (a blocking caller waits for each command before issuing its next, so stripes cannot reorder its stream). The async ordered mode keeps a single stripe — with several stripes a caller's consecutive commands could land on opposite sides of an in-progress drain and execute out of order. Stripes are padded so neighbouring mutexes do not share a cache line. Measured (14-core, in-process throughput bench): windowed GET 5.2M -> 6.0M ops/sec; blocking mixed load 0.60M -> 0.77M ops/sec. External-process harness shows +2-7% with lower p99 (client CPU is not the bottleneck there). * fix(autopipeline): resolve pre-ship review findings in the engine Review of the feature (five independent passes; the two ordering defects were reproduced against a live server) surfaced correctness holes. Engine fixes: - Do executes on a NORMAL connection, outside the pipeline. Raw command names can carry connection state (SELECT/MULTI/SUBSCRIBE/CLIENT ...) or block the connection (BLPOP ...); batched onto a shared pipeline conn they silently poisoned it for every later batch, or stalled and retried unrelated commands. Do now carries plain Client.Do semantics (documented); both faces keep their call shape (blocking waits; async defers via a ready channel). Zero-arg Do reports a real error instead of ErrClosed. - Submit is rejected on the blocking face. Submit does not wait, so a windowed caller defeats the invariant the blocking face's enqueue striping relies on ("every caller waits per command") — reproduced as a deterministic reorder of 25% of adjacent submissions from a single goroutine. The deferred face is the intended home for windowed submission and is unchanged. - NumShards > 1 without Unordered fails construction on the deferred face: round-robin sharding flushes concurrently and does not preserve submit order (reproduced: 42% of windowed pairs reordered at NumShards=4). The blocking face is exempt (callers wait; Submit rejected), as is cluster slot sharding via the new unexported contentSharded marker (same-key commands share a shard, so per-key order holds). - Close no longer errors already-accepted commands: batch execution and permit waits run on a background context (wire timeouts still bound them; Close waits via wg/batchWg before pools tear down), and the shutdown drain now acquires the shard permit so a drained command cannot execute concurrently with — and be observed out of order against — a still-in-flight batch. - Validate rejects AdaptiveDelay with MaxFlushDelay <= 0, which silently disabled batch accumulation entirely while reading as a smart-tuning knob. - numAutoPipelineShards (the cluster wiring default) no longer derives from MaxConcurrentBatches: that coupling collapsed cluster slot routing to a single shard at the default budget, making per-node sharding dead code. - Minors: apStripe padded to a 128-byte stride (the 88-byte stride false-shared hot fields with the next stripe's mutex), 32-bit-safe unsigned modulo in the shard pick (int conversion panicked after counter wrap on GOARCH=386/arm), zero AutoFuture.Wait returns an error instead of nil-panicking, stale "maximizes latency" wording corrected. * fix(pool): pipeline pool survives clone; maintnotifications retires it too Two integration gaps from the pre-ship review: - baseClient.clone() dropped pipelinePool, so a WithTimeout clone silently fell back to the main pool: configured pipeline buffers and pool sizing evaporated with no error, and pipeline traffic polluted main-pool metrics. - The maintnotifications runtime downgrade retired tracked connections only through the primary pool; pipeline-pool connections (tracked identically via initConn) stayed in service with the feature enabled but no hook attached, silently dropping MOVING/MIGRATING pushes. Retire now offers the snapshot to every pool the manager attached a hook to — RetireConns skips connections a pool does not own, so this is safe by construction. * polish(autopipeline): pre-ship review polish — accessors, WaitContext, tidy From the deep review (improvements/docs/dead-code passes): - AutoFuture.WaitContext: ctx-aware wait (abandons the wait, never the command — per-command contexts are still not honored after enqueue, and now there is an escape hatch for callers that need one). - IsBlocking() and Config() accessors: code handed an *AutoPipeliner can tell which face it holds and what the effective (default-filled) config is, instead of probing Submit for an error. - Operator breadcrumb when the 30s permit backstop expires (both the flush and shutdown paths) — previously the one "should never happen" wedge case failed batches silently, with no internal.Logger call anywhere in the engine. - Static errors are errors.New sentinels (errSubmitBlockingFace, errZeroAutoFuture, errDoNoArgs) per house style. - Micro: putQueueSlice zeroes only the used prefix (tail invariant documented; was memclr'ing up to 8 KB per flush for small batches); flushBatchSlice skips provably-empty stripes without locking (safe only in that path — the wake protocol re-observes racing enqueues; the shutdown drain keeps locking). - Docs: the AutoPipeliner type doc's flush-trigger list rendered broken in godoc and implied a lone command never flushes — rewritten, plus a retry caveat (a batch is retried whole on a network error, so non-idempotent commands can execute twice on a dropped connection) and a pointer that Do runs outside the pipeline. Stale backstop comment (claimed Close cancels the permit wait; it deliberately does not) corrected. MaxFlushDelay doc no longer references an unexported identifier. - Dead code: commented-out process() stub removed; unused FIFOSemaphore methods removed (AcquireBlocking, Len, and Close — which would have panicked Release had anyone called it); defaultAccumulateGap is a const; closedBatch renamed completedBatch (it also signals successful direct execution, not just closure). * docs(autopipeline): WaitContext — result may become available later Spell out the post-deadline contract: a ctx error means the wait was abandoned, not the command — the batch is still in flight and may populate the command at any moment. Reading Cmd()'s value or error directly at that point races the executing batch; Wait again and only read after a non-context-error return. * perf(autopipeline): RTT-adaptive coalescing on the default window On a real network the fixed coalescing constants fragment batches. Measured against a Redis Enterprise QA endpoint (RTT ~52ms), 512 workers: when a batch's replies land, hundreds of goroutines wake over several milliseconds, but the flusher flushed 20µs after the first re-enqueues — the herd split into partial batches whose serialization costs a full RTT per split (blocking p50 174ms ≈ 3.3 RTT; manually setting MaxFlushDelay=5ms recovered 7.6k ops/s @ 67ms from 3.4k @ 174ms). On loopback the same constants are exactly right, so no fixed value serves both. The engine executes every batch, so it already knows the round trip: keep an EWMA (alpha 1/8) of batch execution time and derive the default-window timings from it — gap = clamp(RTT/64, 20µs, 3ms) window = clamp(RTT/4, 200µs, 15ms) Loopback RTTs clamp to the historical 20µs/200µs (behavior bit-identical); WAN RTTs widen both so reply-wakeup herds coalesce into one batch per round trip, automatically doing what a hand-tuned MaxFlushDelay did. Explicit MaxFlushDelay / AdaptiveDelay configurations are untouched. The derivation is a pure function (adaptiveCoalesce) for unit testing; EWMA updates are deliberately racy read-modify-writes (losing a sample is harmless, keeps the batch path CAS-free). * perf(autopipeline): merge reply cohorts after a blocked permit wait At MaxConcurrentBatches=1 the flusher takes the queue and THEN blocks on the permit for a full batch round trip. Callers whose replies land just after the take re-submit into the fresh queue during that wait; executing without them splits the herd into two alternating cohorts — each observing two round trips at half throughput — and the state is stable once entered. Measured on a 52ms link: p50 pinned at 120ms (2xRTT) for entire runs at 64 workers x inflight 10 while neighbouring worker counts ran at 1xRTT, and run-to-run bistability at other counts (whichever cohort structure forms at startup persists). After a blocked permit acquire on the default window, debounce the resubmission herd (accumulateBatch's RTT-adaptive gap) and fold the late arrivals into the already-taken batch before executing. The herd collapses back to one batch per round trip. TryAcquire successes (uncontended, or MaxConcurrentBatches>1 with free permits) skip the top-up entirely, and explicit MaxFlushDelay/AdaptiveDelay configs keep their own timing. * perf(autopipeline): replace debounce timer with cohort-tracked coalescing The default coalescing path armed a ~20us debounce timer on every flush. A Go timer that short fires ~1ms late on an idle or low-core host (wakeup latency dominates the requested delay), taxing every low-concurrency command ~5x its round trip: on an in-network endpoint (0.25ms RTT) a lone caller measured p50 1.31ms vs 0.24ms for a plain client. Replace wall-clock guessing with two facts the engine already has: - resubmitDebt: a completed batch of N commands wakes its N waiters together; in a closed loop they resubmit. Credit N at batch completion, debit 1 per enqueue, and hold the flusher until the debt drains - the herd flushes as one deep pipeline, exiting the moment it has landed. An exact per-herd count cannot ratchet into fragmentation the way a smoothed estimate does (an EWMA exit undershoots its own herd and locks into fragmented flushes; measured on a 57ms WAN link). - inFlight: executing batches mean more wakeups are coming. Hold stragglers while anything is in flight (gap-bounded, and below coalesceMinFlush hold them for the next herd rather than burning a pipeline connection on a near-empty flush); when nothing is in flight and no debt is outstanding, the shard is idle and flushes immediately with no timer armed. The silence-gap fallback (200us floor, execEWMA/8, 5ms ceil) only ends waits for callers that left; it never fires in steady state. Also dispatch the single-command fast path in its own goroutine: running Process inline blocked the flusher for a full round trip, and on a slow link a solo straggler then held a landed herd hostage for one RTT whose flush delayed the straggler again - a stable phase-lock at exactly 2x RTT (~25% of runs on a 57ms link). Measured (get=80/set=20, 64B, unordered, MaxConcurrentBatches 50): - loopback: w1 p50 0.156->0.130ms; w64 +16% tp; w256 +14%; w1024 even - in-network 2-vCPU (0.25ms RTT): w1 p50 1.31->0.235ms, 4.7x tp; w16 3.2x tp; w64 2.3x tp; w1024 +4%; w256 -20% (fast-link partial-batch overlap the old timer bought; accepted, fixing it re-fragments WAN) - WAN 57ms: w64/w256/w1024 within baseline variance, phase-lock gone The stops-growing/adaptive-window tests are superseded by autopipeline_coalesce_test.go; autopipeline_stops_growing_test.go and autopipeline_adaptive_rtt_test.go (pipeliner branch) test removed symbols and are replaced by it on rebase. * fix(autopipeline): bound the straggler hold; never erase a fresh herd's debt Two review findings on the cohort coalescing: - The gap-fire stale-debt reset was a blind Store(0) on the pipeliner-global counter. A batch completing on another shard (or racing the fire) between the flusher's last look and the reset would have its just-credited herd erased, fragmenting a herd that is really coming. Reset now requires the debt not to have grown since the timer was last armed, and uses CompareAndSwap so a credit racing the reset itself also survives. - The straggler hold (gap-fire with a near-empty queue while batches are in flight) re-armed unboundedly. With read timeouts disabled a wedged batch pins inFlight forever and up to coalesceMinFlush-1 queued commands would hang with it. The hold is now bounded by autoPipelinePermitBackstop, mirroring the permit wait's safety net. Also drop a stale test comment claiming Options.AutoPipelineConfig is not consulted by the blocking face (it is, when no config argument is given). * refactor(autopipeline): rename debt/herd vocabulary to expected arrivals/wave resubmitDebt reads as if something is owed; nothing is — the counter is an expectation: how many commands the engine knows are about to arrive because it just woke that many waiters. And 'herd' named the woken group by analogy rather than by mechanism. Rename for the next reader: resubmitDebt -> expectedArrivals (announce on completion, account coalesceCohort -> awaitExpectedArrivals per enqueue) cohortGap -> silenceGap (+Floor/Ceil consts) herdExpected -> expectingWave herd / cohort -> wave (comments; 'wave merge' at the permit site) Behavior unchanged; comments rewritten in the same vocabulary. * perf(autopipeline): pad stripes with x/sys/cpu.CacheLinePad The stripe pad was a hand-tuned [80]byte to reach a 128-byte stride. Use golang.org/x/sys/cpu.CacheLinePad instead: it is sized per GOARCH (64 B on x86-64/arm64, 128 B on ppc64, 256 B on s390x), so stripes land on their own cache line on every target rather than a constant that happens to fit the common ones and still false-shares on s390x. x/sys was already in the module graph (indirect); this promotes it to a direct dependency. A contended microbenchmark (8 goroutines, adjacent stripes) shows padded vs unpadded ~16x, and CacheLinePad vs the old [80]byte within noise on arm64 — so no regression, just arch-correct. * perf(autopipeline): lower silence-gap ceiling from 5ms to 2ms * chore(examples,extra): go mod tidy for golang.org/x/sys indirect dep * fix(autopipeline): filter the pipeline pool on Sentinel failover onFailover only filtered connPool, so after a Sentinel promotes a new master the dedicated pipeline pool kept idle connections dialed to the old master and pipelined traffic could keep using them. Filter pipelinePool with the same predicate. Also reword the TestPipelineBufferSizes comment: it verifies end-to-end pipelining with the buffer options set, not the actual socket buffer sizes (which are not observable here). * ci(autopipeline): govulncheck on go 1.26.5 (GO-2026-5856) The govulncheck workflow pinned go-version "1.26.x", which resolves to 1.26.4 and reports GO-2026-5856 (crypto/tls). Pin 1.26.5 (the fixed release), matching the pipeliner/benchmarks branches. * feat(autopipeline): #3 wire into Client and ClusterClient (#3867) * feat(autopipeline): wire AutoPipeliner into Client and ClusterClient Add Client.AutoPipeline() and ClusterClient.AutoPipeline() returning a deferred *AutoPipeliner, plus WithAutoPipeline() and propagation of the pipeline-pool options. Options gains AutoPipelineConfig. * test(autopipeline): cover correctness, ordering, config and cluster routing Standalone and cluster tests for the deferred AutoPipeline API: per-goroutine ordering, no cross-talk or lost commands, error isolation, config validation (panic on MaxConcurrentBatches>1 without Unordered), zero-copy GetToBuffer/SetFromBuffer, adaptive delay, and cluster cross-slot routing. * docs(autopipeline): add README section and 4-way decision-guide example README documents the deferred AutoPipeline API and the ordered-default / Unordered-for-parallel rule. example/autopipeline benchmarks four usage styles (normal blocking, ordered blocking-read, ordered read-later, unordered read-later) and prints a decision guide with throughput and ordering. * feat(autopipeline): AutoPipeline/AsyncAutoPipeline return an error The four AutoPipeline / AsyncAutoPipeline methods (Client and ClusterClient) now return (*AutoPipeliner, error) instead of *AutoPipeliner. These run from post-init calls, so an invalid config must surface as a returned error the caller can handle rather than a panic. On error no instance is cached, so a later call with a valid config still succeeds. Tests updated to the two-value form; the unsafe-config test now asserts an error instead of a panic. * test(autopipeline): drop test for removed AutoPipeline() self-accessor The (*AutoPipeliner).AutoPipeline() stub was removed (orphan, satisfied no interface); drop the test that asserted it returns itself. * test(autopipeline): use AsyncAutoPipeline after unexporting the constructor NewAutoPipeliner is now unexported; the unsafe-config test calls c.AsyncAutoPipeline (same error-on-bad-config contract). Update a stale doc comment reference too. * style: gofmt redis.go and osscluster.go Remove a stray double blank line after the AsyncAutoPipeline methods so golangci-lint's gofmt check passes. * ci(spellcheck): add autopipelining, goroutine(s) to the wordlist The README autopipeline section uses 'autopipelining' and 'goroutine(s)', which aren't in the spellcheck dictionary, failing check-spelling on the PR. Add them to .github/wordlist.txt. * test(autopipeline): fix CI failures on the pipeliner PR Three independent test-job failures, all in test code / the example module: - Blocking-face ginkgo specs panicked: ap.Do returns a generic *redis.Cmd, but the specs type-asserted *StringSliceCmd/*StringCmd. Use *redis.Cmd with its StringSlice()/Text() accessors. - Cluster autopipeline tests failed in the standalone test-ce job: the :7000 tests had no skip guard, and the :16600 tests guarded only on Ping (which passes against unjoined nodes that then return CLUSTERDOWN/MOVED). Add a shared skipIfClusterUnhealthy that gates on CLUSTER INFO cluster_state:ok, and use it in all cluster autopipeline tests. - example/autopipeline didn't compile (govulncheck 'all modules' load error): it called AutoPipeline()/AsyncAutoPipeline() in single-value context after they gained the error return. Assign (ap, err) and panic on err. * ci(spellcheck): add async, runnable to the wordlist The README autopipeline section uses 'async' and 'runnable' in prose; add them to .github/wordlist.txt. Verified the full README passes pyspelling via the exact CI image (jonasbn/github-action-spellcheck:0.62.0). * docs(readme): autopipeline examples use the (ap, err) return AutoPipeline()/AsyncAutoPipeline() return (*AutoPipeliner, error); the README snippets showed single-value calls that wouldn't compile. Capture and check err. Reported by Copilot on PR #3867. * test(autopipeline): cover stops-growing early exit and explicit-delay gating Two white-box tests for the default-window stops-growing behavior: - StopsGrowingDefaultWindow enlarges defaultAccumulateWindow to 1s and asserts a lone caller still flushes in well under it (deterministic, no sub-ms timing). - ExplicitDelayWaitsFullWindow guards the gating: an explicit MaxFlushDelay is still waited in full, so stops-growing never short-circuits a configured window. Config is passed as an argument to AutoPipeline (the blocking face does not read Options.AutoPipelineConfig). * docs(autopipeline): AutoPipeline default is a single ordered stream, not parallel batches * feat(autopipeline): cluster keeps multi-shard default; cover NumShards The engine now defaults to a single shard (deep queue) for standalone clients. Cluster clients still want several shards — commands are routed to shards by slot so each batch stays on one node — so the cluster wiring fills in the numAutoPipelineShards default when the user leaves NumShards at 0, copying the config rather than mutating the caller's struct. Tests cover: standalone defaulting to one shard regardless of the permit budget, an explicit NumShards override, negative-NumShards validation, and the cluster default (copy semantics + explicit passthrough). * test(autopipeline): adaptive-delay test pokes stripe 0 after enqueue striping The white-box adaptive-delay test set the shard's queueLen directly; that counter now lives on the shard's stripes. Point it at stripe 0 (these configs are ordered, so the shard has exactly one stripe). * fix(autopipeline): cluster review fixes — push drain, slot-shard default, docs Cluster-side resolutions from the pre-ship review: - ClusterClient.pipelineReadCmds drains buffered RESP3 push notifications before reading each reply, matching the standalone pipeline and cluster TxPipeline read loops. It was the only push-blind reader: a push frame (e.g. a maintnotifications MOVING notification) arriving mid-batch was consumed as a command's reply, shifting every subsequent reply in the pipeline by one command — and the autopipeliner routes all cluster traffic through this loop. - Cluster wiring uses the decoupled numAutoPipelineShards() default and marks the config contentSharded, restoring per-node slot sharding at the default configs (it had collapsed to one shard, making installAutoPipelineSharding dead code) while remaining exempt from the async NumShards ordering check — slot routing keeps every key on one shard, so per-key order holds. - Docs: ClusterClient.AutoPipeline's godoc was fused into the unexported clusterAutoPipelineConfig helper (the public method rendered undocumented); split and given a per-key ordering qualifier. Stale wording fixed in redis.go ("maximizes latency"), the example ("parallel-batch config"), and command.go (references to never-shipped FutureAutoPipeline/ChanAutoPipeline). - Tests: Submit rejected on the blocking face; NumShards>1 requires Unordered on the async face (blocking and cluster slot-sharded configs exempt); AdaptiveDelay without MaxFlushDelay rejected; Do bypasses batches (a MULTI via Do can no longer corrupt co-batched commands); cluster default config keeps multiple slot-routed shards; stops-growing tests use typed calls now that Do no longer enters the engine. * polish(autopipeline): unify getters, closed-client guard, config fallback, docs Wiring polish from the deep review: - The four AutoPipeline/AsyncAutoPipeline getters (Client and ClusterClient) shared a hand-rolled 15-line caching protocol each; they now delegate to one getOrCreateAutoPipeliner helper, which is also where the new closed-client guard lives: calling a getter after Close returns ErrClosed instead of silently building a fresh pipeliner against closed pools whose flusher goroutines nothing would ever stop. - Options.AutoPipelineConfig now feeds BOTH faces (it was silently ignored by the blocking AutoPipeline — a generically-named option honored by only one of two identically-shaped methods). Each face still falls back to its own default when the option is nil. FailoverOptions gains the AutoPipelineConfig field (failover clients could reach the feature but not configure it) and plumbs it to both the sentinel-backed client and the failover cluster client. - Docs: getter godocs state the Options fallback; ClusterClient.AsyncAutoPipeline documents per-key ordering across nodes; the README's stale "parallel pipelines" claim is corrected to the ordered-stream default, throughput figures are qualified, and the section gains the connection-count benefit plus a caveats paragraph (per-command contexts, blocking commands and Do bypass batching, whole-batch retry duplication, shared cached instance). Buffer-size option docs mention both pipeline entry points. - The scoped "unused" linter exemption for the stacked-PR period is removed: the wiring lands on this branch, so every engine symbol is referenced. Example build artifact gitignored. * docs(example): autopipeline example becomes a usage tour + benchmark The example was a throughput harness only — valuable, but it never showed how to actually use the API (nothing read a value back). It is now two acts: Act 1, a usage tour: the blocking face as a drop-in (goroutines Set+Get with plain-client call shape), the async face's submit-a-window-read-later pattern, Submit/AutoFuture for raw commands (async-only by design, with WaitContext noted), the Do escape hatch (runs on a normal connection outside the pipeline — plain Client.Do semantics), in-code tuning notes (Unordered + MaxConcurrentBatches 2-4, NumShards stays 0, first-call-config-wins caching, dedicated pipeline pool options), and an env-gated cluster tour (REDIS_CLUSTER_ADDRS) showing slot sharding needs no extra code. Act 2, the existing four-way throughput comparison, with review fixes: the unused rdb parameter dropped, the bench-tuned MaxConcurrentBatches: 80 magic number replaced by the recommended 2-4 (with a comment where big numbers belong), the baseline's PoolSize asymmetry explained where it stands, panic replaced by the file's print-and-exit convention, and the stale "parallel pipelines" package comment corrected to the ordered-stream default. Adds the README the example was missing (15 of 17 examples have one): run instructions, what each act shows, a sample results table, and the caveats users must know (per-command contexts, whole-batch retry duplication, per-key cluster ordering). * test(autopipeline): re-point tests that the Do change made vacuous AutoPipeliner.Do now executes on a normal connection outside the engine, so every test that drove ap.Do to exercise batching/flushing/queueing was passing vacuously against the plain-client path. Re-pointed at the engine: - ap.Do(ctx, "SET", ...) becomes the typed ap.Set(...) throughout the ginkgo suite, TestAutoPipelineBasic/MaxFlushDelay/Concurrency/SequentialSingleThread, the sequential tests, and the cluster tests (which now genuinely exercise slot sharding). - The error-isolation spec uses a typed WRONGTYPE failure inside a batch instead of Do("INVALID_COMMAND"), which no longer touches a batch at all. - The queue-length spec moves to the deferred face with a wide explicit window, making Len assertions deterministic and meaningful (they previously observed an untouched queue: always zero). - TestAutoPipelineSecondaryAccessors moves to the deferred face — the missing await() regression it guards is only expressible there (the blocking face waits inside the call, so it could never catch it). Stale reference to the removed autoPipelineCmd type dropped. - The sequential submit-then-read tests move to AsyncAutoPipeline (that is the deferred pattern they describe); TestOrderedModeWindowed likewise, making its "never blocks between submits" claim true. TestFutureFaceOrdering deleted — it duplicated TestBlockingFaceExecutesOnCall. Note: the ginkgo AutoPipeline specs require the full test rig (BeforeSuite spins up cluster/sentinel processes) and are validated in CI; everything reachable locally runs green under -race. * docs(readme): list Automatic pipelining under Features (experimental) Adds the feature-list entry linking to the section, marked experimental and scoped to its intended audience: high-throughput / high-load / scale use cases. The section itself now carries the same experimental note so readers landing via the anchor see it too. * docs(readme): point to the example app from the autopipelining intro The example pointer sat at the end of the caveats paragraph where nobody scrolling the section would find it; it now lives in the section's opening note (and the duplicate tail mention is removed). * test(autopipeline): pin the RTT-adaptive coalescing derivation Unit tests for adaptiveCoalesce (loopback clamps to the historical constants, WAN scales gap/window, ceilings bound the tax, enlarged-floor interaction with the stops-growing test hook) and the execEWMA smoothing behavior. * test(autopipeline): drop tests of removed coalescing internals autopipeline_stops_growing_test.go and autopipeline_adaptive_rtt_test.go exercised defaultAccumulateWindow / adaptiveCoalesce, which no longer exist; autopipeline_coalesce_test.go covers the replacement mechanism. * fix(autopipeline): non-pipelineable command fails itself, not the merged batch mapCmdsByNode returned a batch-wide error when any command's policy failed CanBeUsedInPipeline, so under the opt-in dynamic resolver one caller enqueueing e.g. a multi-shard MGET errored every unrelated caller co-batched in the same autopipeline flush (review finding M9). Set the error on the offending command and continue mapping the rest; the skipped command never joins a node pipeline, is not retried, and its waiter reads the policy error while everyone else proceeds. * docs(example): autopipeline example catches up with the engine - Do guidance: typed blocking commands (BLPop, XRead Block, Wait, ...) are diverted to a normal connection automatically — Do is not the escape hatch for them; and stop suggesting SELECT/SUBSCRIBE through Do (connection-state commands poison a pooled conn with plain Do semantics too — Options.DB and the PubSub API are the answers). - Tuning notes: extra MaxConcurrentBatches permits add overlapping batches bounded by the pipeline pool; with wave-boundary coalescing they no longer fragment batches. Reworded in main.go and the bench comment. - README: refresh sample numbers on the current engine (blocking face 552k -> 633k on the same loopback box) and note the pipeline-hook instrumentation shape in caveats. * ci(autopipeline): wordlist loopback/autopipeliner; go mod tidy example/autopipeline * chore: stop tracking .claude/worktrees gitlinks; gitignore them * perf(autopipeline): dispatch drained batches straight to processPipelineHook The flusher built a Pipeline per batch: one Pipeline alloc, a nil cmds slice growth-reallocating ~log2(N) times, and one interface Process call per command — then Exec handed the slice away, nothing reused. Extend cmdableClient with processPipelineHook (the exact hook-wrapped entry Pipeline.Exec wires, so hooks/OTel see the identical call) and hand the drained queue slices to it directly: zero-copy for single-stripe drains (every ordered shard), one pooled merge for striped drains. Shutdown flush uses the same path. Local A/B (M4 Max, redis 6379, count=6): B/op -12-14% on windowed workloads (p=0.002); blocking-face throughput +10% (1.09M -> 1.21M ops/s); windowed throughput unchanged (RTT-bound). Note: hooks now receive a pooled slice — retaining []Cmder past the hook call was never sanctioned; contents are zeroed on reuse. * perf(autopipeline): skip the shard round-robin atomic when there is one shard len(shards)==1 is the standalone default; every enqueue still bumped ap.next — a shared cache line — for a pick that is constant. Same guard the stripe pick already has. Also adds a goroutine-leak regression test (open/close many autopipelines, assert goroutines return to baseline). Local A/B: unordered dispatch ~-3% ns/op, windowed throughput +3-5%. * test(autopipeline): de-flake TestDoBypassesPipeline Force a real multi-command batch (MaxFlushDelay) so the batched SETs run on the dedicated pipeline pool, instead of racing the single-command fast path onto the MULTI-poisoned normal-pool connection. Fixes intermittent failures under -race and on loaded CI runners. * fix(autopipeline): surface retry-exhaustion errors and route streaming re-auth to the owning pool generalProcessPipeline only set command errors on the early-exit branch, not when the retry loop was exhausted. Manual Pipeline surfaces that via Exec's return value, but AutoPipeline discards the pipeline error and observes results only per-command, so a network failure that exhausted retries silently returned a nil error and zero values. Mirror the early-exit setCmdsErr on exhaustion, guarded by !isRedisError so a per-command redis error keeps its own reply. Streaming-credentials re-auth hardcoded c.connPool even for pipeline-pool connections, so a failed pipeline conn was closed and accounted against the main pool. Route the SingleConnPool base and the auth-error CloseConn to the owning pool via poolForConn (pool.Conn.PoolName). * test(autopipeline): retries, cluster MOVED, and pipeline-pool feature coverage - pipeline and autopipeline retry re-drive: success-after-retry, exhaustion surfaces the error, WRONGTYPE isolation, context-cancel not retried (scriptable bad-conn dialer, no external deps) - cluster MOVED redirect inherited by AutoPipeline via the shared execer, with a MaxRedirects=-1 anchor proving the redirect path is exercised - dedicated pipeline pool: ACL AUTH + SELECT handshake, StreamingCredentialsProvider re-auth, and maintnotifications all verified on pipeline-pool connections * feat(autopipeline): expose AutoPipeline on the UniversalClient interface Add AutoPipeline/AsyncAutoPipeline to the UniversalClient interface. *Client (including sentinel-backed failover clients) and *ClusterClient already implement them; *Ring returns errRingAutoPipelineUnsupported so it satisfies the interface and fails explicitly instead of being silently unavailable. * test(autopipeline): context contract, reply-shape demux, fuzz, backpressure, UniversalClient - AutoPipeline context contract: WaitContext abandons only the wait, caller ctx does not cancel execution, blocking commands honor ctx, submit-after-Close - heterogeneous reply-shape demux (bulk/array/int/map/nil/double/large/binary) under RESP3 and RESP2 - differential fuzz: identical results plain client vs autopipeline over random type-diverse command streams - blocking-command isolation, double-Close, windowed backpressure, bounded soak - AutoPipeline reachable via the UniversalClient interface (Ring returns unsupported) * chore(autopipeline): fix CI — remove unused putPipeliner, pin govulncheck to go 1.26.5 - lint: putPipeliner (pipeline.go) was documented but never wired to a caller, so golangci-lint flagged it (unused); remove it. - govulncheck: pin the scan's Go toolchain to 1.26.5, fixing the crypto/tls standard-library advisory GO-2026-5856 (was resolving to 1.26.4). * feat(autopipeline): AutoPipeliner implements UniversalClient (drop-in) Widen the internal cmdableClient to UniversalClient and delegate the non-batched surface (AddHook, Watch, Subscribe/PSubscribe/SSubscribe, PoolStats, Do, AutoPipeline/AsyncAutoPipeline) to the underlying client, so an *AutoPipeliner can be used anywhere a redis.UniversalClient is expected: data commands are batched, everything else runs on the underlying client. Narrow Do to return *Cmd to match the interface; drop the now-redundant .(*redis.Cmd) assertions in the tests. Close() still closes only the autopipeliner, not the underlying client. This lets tests (the command-suite matrix and maintnotifications e2e) parametrize a single UniversalClient subject that includes the autopipeliner. * refactor(autopipeline): take *AutoPipelineConfig (nil=default) not variadic Address review feedback (cxljs, #3867). The variadic `config ...*AutoPipelineConfig` reads like the functional-options pattern but isn't; a single `*AutoPipelineConfig` where nil selects the default is clearer. Changes AutoPipeline / AsyncAutoPipeline on *Client, *ClusterClient, *Ring, the UniversalClient interface and the AutoPipeliner drop-in, plus getOrCreateAutoPipeliner's override parameter. Call sites now pass nil instead of no argument. README, example and doc comments updated. * fix(autopipeline): PR #3867 review fixes - clone(): copy pipelinePoolName so timeout-derived clients keep the dedicated pipeline pool for re-auth routing (poolForConn) instead of falling back to the main pool. - clusterAutoPipelineConfig: mark contentSharded even when the caller sets an explicit NumShards. A cluster always routes by slot, so per-key order holds regardless of shard count; the async face was previously rejected. - TestFutureFaceTyped: build the deferred face with AsyncAutoPipeline so it actually exercises await()/deferred accessors. - Options docs (sentinel/ring/osscluster): correct the pipeline-pool enable condition — set by PipelineReadBufferSize/PipelineWriteBufferSize; PipelinePoolSize alone does not enable it. - README: define ctx in the AutoPipeline examples so they compile as-is. - Update the cluster shard-config test for the corrected contentSharded behavior. * refactor(autopipeline): AutoPipelineOptions + no-arg / WithOptions methods Rename the config type AutoPipelineConfig -> AutoPipelineOptions (matching the Options naming family: Options, ClusterOptions, ...) and split each face into a clean no-arg method plus an explicit variant: AutoPipeline() // Options.AutoPipelineOptions or default AutoPipelineWithOptions(*AutoPipelineOptions) AsyncAutoPipeline() AsyncAutoPipelineWithOptions(*AutoPipelineOptions) Drops the awkward AutoPipeline(nil) for the common default case while keeping an explicit per-instance override. Both faces remain independent (a client can hold a blocking and an async autopipeliner at once). Also renames Default{,Blocking}AutoPipelineConfig -> ...Options and the *Options AutoPipelineConfig fields -> AutoPipelineOptions. Applied across *Client, *ClusterClient, *Ring, the UniversalClient interface and the AutoPipeliner drop-in; call sites, README and example updated. * docs(autopipeline): reword no-arg -> no-argument (check-spelling) check-spelling flagged "arg" (tokenized from "no-arg") in the README. Use "no-argument" instead. * fix(autopipeline): hooks reading deferred results no longer deadlock A pipeline (or process) hook that reads a deferred command's result — cmd.Err()/Val()/Result() — blocked on the batch's done channel, which the dispatching goroutine closes only after the hook chain returns: a deterministic self-deadlock on the async face (#3867 review). Any hook that inspects per-command results inside the chain — a common pattern in custom logging/metrics instrumentation — triggers it. The blocking face was unaffected. Fix, leaving the execution path untouched: - Complete batches at the innermost exec seam. The async dispatch now runs the user-hook chain via withProcessPipelineHook around the base processPipeline, closing the batches (and announcing expected arrivals, preserving the wave-merge ordering) the moment replies are populated — still inside the chain, so a hook reading results after next() (the span-status/metrics pattern) finds them ready. The lone-command fast path and Do get the same treatment through withProcessHook. The flusher's deferred close stays as a panic backstop (apBatch.close is now idempotent). The blocking face keeps its previous flow unchanged. - Guard await() against self-deadlock for reads BEFORE next(): the batch records its dispatcher's goroutine id while inside the hook chain; a reader on that goroutine gets the not-yet-executed view — exactly what a plain pipeline hook sees before next() — instead of blocking on a channel only its own goroutine can close. The goroutine id is parsed only on the about-to-block slow path, never on the fast path. - Make cmd.ready an atomic pointer: processAsync wired it after the command was already published to the flusher, racing a hook's read. The atomic store keeps that race benign — a hook that wins sees nil and takes the non-blocking not-yet-executed view — with nothing added under the hot stripe enqueue lock. - Arm the goroutine-id guard only when user hooks are installed (hooksMixin.hookCount): hook-free clients never stamp the batch, so a blocked reader pays one atomic load, not a runtime.Stack parse. Benchmarked against the unfixed base (benchstat; sequential n=8 rounds plus drift-controlled interleaved runs for the sensitive benches): DispatchPath ordered/unordered/solo, Submit, FutureFace and Throughput are all statistically unchanged (~, p>0.05); allocs identical (4/op dispatch). An earlier iteration that put setReady under the hot stripe enqueue lock benched at -9% Submit ops/sec and was rejected; the shipped shape adds nothing under that lock and keeps the guard to one atomic load unless hooks are installed. Note: results become readable the moment execution completes, so a hook that mutates a command's result after next() is racy on the async face; hooks should treat results as read-only there. Regression tests cover before/after-next hooks across the batch, solo and Do dispatch paths plus the blocking face, race-enabled; the deadlock reproduces on the unfixed tree. * test(autopipeline): consolidate per-topic test files into two Collapse the 29 autopipeline_* / osscluster_autopipeline_* test files into: autopipeline_test.go all blackbox tests (package redis_test), including the cluster autopipeline tests autopipeline_internal_test.go whitebox tests (package redis): shard-count decoupling, coalescing, review-fix regressions Pure reorganization: no test bodies changed, only regrouped (matches the consolidation already done on the benchmarks branch). Counts preserved (68 Test funcs, 3 Describe suites). The pipeline_*_test.go files are left as-is: they cover the dedicated pipeline pool and plain-pipeline behavior, not the autopipeline engine. * fix(autopipeline): deep-review fixes — hook short-circuit, cluster guard, tests, docs Engine: - A hook that returns WITHOUT calling next (allowed by the Hook contract; circuit-breaker shape) used to wake every waiter with a silent zero value: the chain's error was discarded on all dispatch paths. The innermost now records execution; when a hook short-circuits, its error (or a dedicated sentinel) is set on every command before the batch completes. Covers both faces, the lone-command fast path, Do and the shutdown drain. - The async self-deadlock guard is now always armed on cluster clients: node clients carry their own hooks (OnNewNode + AddHook, the redisotel pattern) that hookCount() cannot see. Node-level hooks run on node-worker goroutines, which the goroutine-id guard cannot protect pre-next(); that limitation is documented (results are safe to read after next()). - AutoFuture.Wait/WaitContext get the same dispatcher-goroutine guard as the command accessors. - Lone-command dispatch releases its permit and inFlight slot before waking the caller, so a closed-loop lone caller's next command takes the immediate-flush path instead of arming the silence-gap wait. - AddHook mutates the hook slice under hooksMu, and the dispatch snapshots the slice under RLock (background flushers iterate it concurrently). Tests: - TestAutoPipelineConcurrency asserted a locally-incremented counter and could not fail; it now checks every command and verifies DBSize. - The autopipeline reply-demux test ran on the blocking face sequentially, so every command took the lone-command fast path and the batch demux it claims to cover never executed; it now uses the async face with a flush window so all nine heterogeneous commands land in one pipeline batch. - The three cluster autopipeline tests dialed :7000-:7002 and always skipped; they now use the test harness's :16600-:16602. - Plain go tests use apTestAddr() (REDIS_PORT-aware, :6379 default): the suite variable redisAddr defaults to :6380 and is only corrected inside the ginkgo BeforeSuite, which plain go tests may run without. - Cold-start latency tests warm up dial+handshake before timing; goroutine leak checks poll with a deadline instead of fixed post-GC sleeps; the flush-on-timer test waits on a completion signal instead of sleeping. - New coverage: hook short-circuit (both faces, batch+solo), hooks added after the autopipeliner exists (client and delegate AddHook), Close flushing pending deferred futures, client-level Options.AutoPipelineOptions taking effect on both faces, BLPOP on the async face (readTimeout diversion), and AdaptiveDelay end-to-end traffic. API/docs: - UniversalOptions gains AutoPipelineOptions, threaded through Simple/ Failover/Cluster (the other option structs already had it). - ErrRingAutoPipelineUnsupported is exported (it was named in public godocs while unexported and unmatchable with errors.Is). - ClusterOptions.AutoPipelineOptions doc corrected: it configures BOTH faces, not only AsyncAutoPipeline. Client/ClusterClient Close docs now mention both cached autopipeliners. - README: stale WithConfig name fixed, the parallel-batches example uses a realistic MaxConcurrentBatches (8) with a note that more permits don't deepen batches; godoc throughput claims hedged as local loopback measurements; stale setReady/await comments refreshed; MaxBatchSize doc notes the blocking face's 300 preset. - Remove three stray benchmark screenshots (513 KB) accidentally committed. * fix(osscluster): route cluster pipelines through the dedicated pipeline pool Cluster pipelines acquired connections via node.Client.getConn — the node's main pool — so the dedicated pipeline pool was never used on ClusterClient, even though node clients inherit PipelineReadBufferSize / PipelineWriteBufferSize / PipelinePoolSize through clientOptions (review finding by cxljs on #3867). processPipelineNode and processTxPipelineNode now acquire through withPipelineConn: the node's dedicated pipeline pool when configured, with the exact previous behavior as the fallback (withPipelineConn delegates to withConn when no pipeline pool exists). An entered flag distinguishes acquisition failures — which keep the old MarkAsFailing / mapCmdsByNode / setCmdsErr handling — from execution errors, which processPipelineNodeConn already handles. TestClusterPipelineUsesDedicatedPool covers both the manual pipeline and the autopipeline batch path against a real cluster and asserts the node pipeline pools actually carried the traffic. * fix(autopipeline): DoRaw escape hatch + node-hook short-circuit surfacing Address review feedback (ofekshenawa, #3867): - AutoPipeliner embeds cmdable, so DoRaw and DoRawWriteTo rode the batching engine even though Do deliberately runs outside it. Raw commands carry Do's caveats, and DoRawWriteTo streams to caller-provided io.Writers — user I/O that must not execute inside a shared batch's reply loop. Both now override with Do's exact escape-hatch shape through a shared runOutsidePipeline helper: synchronous Process on the blocking face; returns-immediately with hook-safe batch completion (and hook short-circuit surfacing) on the deferred face. Do is refactored onto the same helper. - processPipelineNode and processTxPipelineNode discarded the return of the node-level hook chain: a node hook returning without calling next (the circuit-breaker shape the Hook contract allows) left every command error-free, so a cluster pipeline could report success for commands that were never sent. The chain now records execution; when it never reached the transport, the chain's error — or the errHookShortCircuit sentinel — is set on every command. Deliberate aborts are not remapped for retry: a retry would re-run the same hook. Tests cover DoRaw/DoRawWriteTo on both faces under result-reading hooks (including the no-args guard) and the node-level short-circuit against a real cluster, using an armable pipeline-only breaker installed via OnNewNode — disarmed during warmup because connection-init handshakes are themselves pipelines through the hook chain. * fix(autopipeline): retry tests honor REDIS_PORT, doc nits - pipeline_retry_test.go hard-coded ":6379" four times, silently skipping (or hitting the wrong server) when tests run with REDIS_PORT set; use apTestAddr() like the rest of the suite - drop a duplicated sentence in the ClusterOptions doc - the example benchmark's comment promises a goroutine-sized pool for the plain-client baseline but set 50 for 500 goroutines; size the pool to the goroutine count so the baseline is not pool-starved * fix(autopipeline): honor a hook's post-next verdict; complete batches after the chain A hook that called next and then returned an error had that verdict discarded (review finding by ofekshenawa on #3867): batches completed at the innermost exec seam, so waiters were already awake and the executed flag suppressed the chain's error. Batches now complete at the caller's deferred closes, after the whole hook chain has returned. The dispatcher-goroutine-id guard in await()/Wait() carries the self-deadlock protection instead: a hook on the dispatch goroutine reading a command before next() gets the not-yet-executed view, and after next() the populated results — without blocking on a channel only its own goroutine completes. This restores plain-pipeline hook semantics across the board: - a post-next injected error (exec succeeded, hook returned an error) reaches every command that has no error of its own; per-command errors recorded by the exec itself always win, and a hook's rewrap of an exec failure is dropped rather than clobbering precise per-command results; - the lone-command fast path and Do record the chain's final verdict exactly like Client.Process — short-circuits, rewrites and suppressions are all honored; - hooks may adjust results before any waiter wakes, removing the previous 'results are racy to mutate after next()' caveat; - the expected-arrivals announce still precedes the closes on both faces, preserving the wave-merge ordering. Also: the differential fuzz passed a raw 1000 (nanoseconds) to Expire, which truncates to EXPIRE 0 (a delete) — now a real 100s TTL (review finding by cursor). New TestAutoPipelineHookPostNextError covers both faces across the batch, solo and Do paths; the full hook battery (deadlock before/after next, short-circuit, hooks-added-late, DoRaw) stays green through the guard, as do the AP race suite and the cluster tests against a live cluster. * chore(examples): tidy the new example modules against this branch's root The master merge added four example modules (ft-aliaslist, himport, search-aggregate-collect, ts-querylabels) tidied against master's root go.mod. This branch's root additionally requires golang.org/x/sys (cache line padding in the pipeline pool), and the examples replace the root module, so on this branch they need 'go mod tidy' — which made the govulncheck CI job fail while loading packages (no vulnerability involved). Tidy adds the x/sys indirect requirement to each. * fix(autopipeline): skip the dispGid stamp on the blocking face's solo path The lone-command fast path stamped the dispatcher goroutine id (a runtime.Stack header parse) whenever hooks were installed, without the !ap.blocking guard the multi-command and shutdown dispatch paths already have. On the blocking face the stamp is never consulted — blocking commands never carry a batch, so their result accessors don't block — making it pure overhead on every single-command flush (review finding by cursor on #3867). Also document, at the verdict guard in dispatchCmds, why a post-next hook error is applied only to an all-clean batch: hooks conventionally return next's error, so after a partial failure the chain error is presumed to be that echo, and stamping it on the commands that did succeed would overwrite valid replies with their batchmates' failure. Behavior unchanged. * fix(autopipeline): never read user command results on the execution path A full re-review of the PR found that the async autopipeliner deadlocked in the DEFAULT configuration. pipelineProcessCmds re-read the first command's error after the himport re-issue pass via Err(); on the async face Err() awaits the command's own batch — which this very goroutine is executing and only closes after the hook chain returns. With no hooks installed the dispatcher-goroutine-id guard is unarmed, so a multi-command batch whose first reply was any error (a plain redis.Nil cache miss is enough) blocked the dispatch goroutine on itself forever: shard permit never released, every waiter hung, Close() hung on the batch WaitGroup. Reproduced with 50 async GETs on missing keys — the runtime reports 'all goroutines are asleep'. The same execution-path Err() reads existed at five himport sites (himportAfterBatch, himportRetryFailedSets, the cluster re-queue), which additionally run on per-node worker goroutines where the guard can structurally never match — any async batch containing an HIMPORT command wedged even when every command succeeded. All six now use rawErr(), matching the already-swept neighboring sites, with comments stating the rule. Two regression tests pin the redis.Nil-first batch and the himport-in-batch cases on the hook-free default. Also from the re-review: - TSNRangePivotRowSliceCmd.Val()/Result() were the only command accessors missing await(): async-face TSNRange/TSNRevRange returned zero values while racing the flusher's readReply. Added; a mechanical sweep of every accessor confirms no other misses. - Client.WithTimeout copied the client struct while AutoPipeline()/Close() can concurrently mutate the mutex-guarded autopipeliner fields — a data race under -race; the snapshot is now taken under the guard. - NewFailoverClient created the dedicated pipeline pool but never set pipelinePoolName, so poolForConn would route failover pipeline conns to the main pool (latent until failover clients gain streaming credentials; wrong-pool CloseConn/accounting once they do). - AutoPipelineOptions coerced non-positive MaxBatchSize and MaxConcurrentBatches to defaults BEFORE Validate ran, so the documented negative-value construction errors were unreachable. Validate now runs first; zero still selects the default. Test-quality fixes (found by auditing what the suite actually pins): - The pipeline/AP 'retries on network error' tests were vacuous: the broken conn they seeded into the idle pool embeds a zero net.TCPConn, which the pool health check discards before use — both tests passed with retries disabled. They now use a dialer whose conn fails exactly one write at the wire after the handshake, and assert the failure was consumed and exactly one redial occurred, so the first attempt genuinely dies and the re-drive is exercised. - The differential fuzz drove the blocking face sequentially, so every command took the lone-command fast path: zero multi-command batches, meaning batch demux had no randomized coverage. Added an async windowed phase (submit all ops, then read — the ordered default preserves submit order, so the sequential oracle still applies) with a counting hook that fails the test if no real batch formed. Validated by mutation: swapping two replies in dispatchCmds now fails the fuzz. - The cluster NumShards default test asserted >= 2, which fails on single-vCPU runners (the default is clamp(GOMAXPROCS,1,16)); it now asserts the computed default. - README: the AutoPipeline getters also return ErrClosed on a closed client, not only invalid-config errors. * fix(autopipeline): async-face parity for metric-only result reads; CI and doc fixes Publish/SPublish and XRead read their command's result only to record OTel metrics. On the async autopipeline face a result read awaits execution, so with no recorder installed these silently turned fire-and-forget publishes into blocking calls. internal/otel gains Enabled() and the three sites are gated on it; TestAsyncAutoPipelinePublishDoesNotAwait pins submission completing inside the flush delay. The Eval family's post-issue read stays: it normalizes NOSCRIPT to ErrNoScript, which Script.Run/RunRO depend on — now documented as synchronous on the async face. The failover client now registers its pools (main, pubsub, pipeline) for OTel gauge metrics, matching NewClient; it previously registered none while still unregistering on close. example/client-side-caching (new with the csc master merge) is tidied against this branch's root module — the govulncheck job failed loading it (go mod tidy needed; no vulnerability involved), the same pattern as the four examples fixed before the rebase. Documented from the cross-cutting review: a hook must never issue a command on the same autopipeliner and synchronously wait for it (the nested command needs the dispatch slot the hook chain holds; the engine recovers only by failing that flush after the 30s permit backstops, taking co-drained bystanders with it); Options.Limiter is consulted once per batch dispatch like a manual pipeline, not once per command; an autopipeliner created on a WithTimeout clone is not stopped by the parent's Close and must be closed explicitly. README caveats plus the armSelfDeadlockGuard and WithTimeout doc comments carry the warnings. * chore(autopipeline): consolidate pipeline test files; mark the feature experimental Consolidate the five loose pipeline test files into two, mirroring the engine PR's test consolidation: the dedicated-pool battery (handshake, maintnotifications, streaming credentials) into pipeline_pool_test.go and the execution-behavior tests (heterogeneous reply-shape demux, the retry/re-drive loop) into pipeline_exec_test.go. Pure move — ten test functions before and after, with section markers keeping provenance. Mark autopipelining EXPERIMENTAL using the repo's existing marker line: the API may still change as feedback lands. The marker covers every public surface — the eight client getters (Client and ClusterClient, blocking and async, no-arg and WithOptions), the AutoPipeliner and AutoPipelineOptions types, both Default*Options constructors, the UniversalClient interface block and Options.AutoPipelineOptions — plus banners in the README autopipelining section and the example. * chore(autopipeline): assert AutoPipeliner's UniversalClient conformance in universal.go The compile-time assertion existed in autopipeline.go; move it into universal.go's assertion block next to Client, ClusterClient and Ring, where the interface and its implementors are declared. * chore: gofmt — drop the double blank line left by the assertion move * chore(autopipeline): skip the fallback options allocation when an override is supplied getOrCreateAutoPipeliner called the fallback constructor (which allocates a default AutoPipelineOptions) before checking for an explicit override, discarding the result whenever one was provided (review finding by cursor on #3867). Check the override first. * fix(autopipeline): run SHUTDOWN and MONITOR outside pipeline batches Both are on the typed command surface and rode pipeline batches. A batched SHUTDOWN kills the server before it replies, so every batchmate failed with EOF and the whole batch was retried against a dead server (late review finding by cxljs on #3865). A batched MONITOR rebinds the shared connection into a monitor stream, desyncing every batchmate reply behind it. Both now execute directly on their own pooled connection, exactly like blocking commands — one gate in submit(), shared by both faces. The rest of the typed surface audits clean: Quit is a deprecated panic stub; the connection-state commands (Auth, Select, Hello, SwapDB, ClientSetName, ClientSetInfo) live on statefulCmdable and are not reachable through the AutoPipeliner; ClientTracking is the same footgun on a plain pooled client (client-side caching is the supported path); CLIENT PAUSE and the BGSAVE/FLUSH family are server-wide and batch-safe; blocking commands were already routed out via their read timeout. TestAutoPipelineShutdownRunsOutsidePipeline pins both routings with path-recording hooks against a guaranteed-closed port, so nothing real is shut down by the test. * fix(osscluster): honor a node-level hook's post-next verdict on clean sub-batches processPipelineNode surfaced a node-level hook's error only for short-circuits (hook never called next). A hook that called next and then returned its own error — the node-level twin of the post-next case fixed in dispatchCmds — had that verdict dropped: on an all-clean sub-batch every command kept reporting success (suppressed Copilot finding on #3867). Apply the same rule as AutoPipeliner.dispatchCmds: a post-next error on a fully clean sub-batch can only be the hook's own verdict, so it is applied to the commands; on a mixed batch the exec-recorded outcomes win (hooks conventionally echo next's error, and stamping the echo would overwrite successful replies). No remap for retry — retrying would re-run the same hook. cmdsFirstErr reads rawErr, so the check cannot await on the node goroutine. TestClusterNodeHookPostNextError pins the behavior against a live cluster with an armable post-next hook (disarmed during warmup so conn-init handshake pipelines pass). Also: example/autopipeline/go.mod pinned v9.21.0 while every other example pins v9.22.0-beta.1 — aligned and tidied. * fix: deflake the release-time push peek and the handoff-removal test TestReleaseConnRemovesConnectionAfterPartialPushRead failed randomly across matrix cells, and the root cause is in product code: peekAndProcessPushNotifications set a 10µs read deadline after MaybeHasData() had already confirmed kernel-buffered bytes. On a loaded machine the deadline routinely expired before the first peek syscall ran; the push processor treats a zero-consumed timeout as 'no pending data', so a connection with buffered push bytes went back into the pool undrained instead of being drained or removed. The first read never waits on the network here — the deadline only needs to survive scheduler pauses — so raise it to 1ms. Reproduced on the base branch (300 plain / 100 -race iterations both fail); the same loops are fully green with the fix. TestConnectionHook/ConnectionRemovalOnHandoffFailure slept a fixed 100ms and then asserted the connection had been removed — racing the handoff worker's three retries with backoff on slow runners. Poll for the removal with a bounded deadline instead. * feat(autopipeline): #4 tests and benchmarks (#3892) * feat(autopipeline): #4 benchmarks and parametrized command-suite runs Rebuilt on the post-#3867 integration: the branch's history carried pre-rebase copies of pipeliner content that conflicted with the squashed merge, so this is the branch's actual payload as one commit. - autopipeline_bench_test.go: eight honest benchmarks — plain-client and manual-pipeline baselines, the dispatch path (per-command ns/op and allocs/op), the future face, Submit, zero-copy reads, cluster throughput, and the plain-vs-blocking-vs-windowed headline comparison — updated to the shipped API. Every fixed-duration driver divides by the timed region (the previous fixed divisor inflated cluster windowed ~2x and the plain baseline ~27%), zero-copy read buffers live outside the timed region so B/op reflects the command path, and every bench skips without a server. The earlier tuning sweeps (batch/flush-delay/buffer sizes) were deleted: at low parallelism they reported the configured MaxFlushDelay timer for every swept value — noise dressed as signal. autopipeline_bench_README.md rewritten multiplier-first: absolute numbers are machine/load-dependent, so the documented signal is the within-run ratio vs the plain-client baseline (plus environment-independent allocs/op); the main README's throughput claims got the same treatment. - iterator.go: ScanIterator.Next read the command's page/cursor fields without awaiting the fetch, which only enqueues on the deferred autopipeline face — iteration spun re-issuing SCANs with a stale cursor forever, flooding the engine with an unbounded batch and hanging the caller. Next now awaits the fetch before reading (a no-op on every other path); TestAsyncAutoPipelineScanIterator pins it. - Parametrized-suite fixes: BFScanDump/CFScanDump compared command structs (each async command carries its own batch pointer, so equality is structurally false there) — compare values; the SlowLog argument-count spec and the BLMoveM timeout spec now keep admin/conn calls on the raw client like their siblings; the commands/bitmap suites build the subject closer before their first assertion so a failed BeforeEach cannot leave AfterEach with a nil or stale closer. - CI: the subjects job focus covers all nine parametrized suites (the old case-sensitive focus silently ran one), each subject run carries an explicit -timeout, and the run-tests action refuses an empty make target. - Command-suite parametrization (#3885): the commands/array/bitmap/ iterator/json/probabilistic/search/timeseries/vectorset suites run against a subject selected by GOREDIS_TEST_SUBJECT (client, ap-blocking, ap-async) via the newUniversalSubject harness, so the whole surface can be replayed through either autopipeliner face without touching specs. - maintnotifications: a failed manager Close no longer drops the not-yet-processed additional pool hooks — they are put back so a retried Close still sees them. - pipeline_buffer_test.go: skip, rather than fail, when the docker test server is absent. - CI: a test-autopipeline-subjects job replays the command suite through both faces via GOREDIS_TEST_SUBJECT (run-tests action gains an injection-safe make-target input; Makefile gains the target). Dropped from the old branch: stale pre-rebase engine-file copies (superseded by #3867), its pre-consolidation AP test files (integration's are newer and complete), and three unreferenced PNGs. * chore(autopipeline): review polish — close leaked test clients, align the docs The ScanIterator regression test closes its autopipeliner explicitly and the RESP3 search suite closes client2 in AfterEach (it leaked one client per spec) — both flagged by Copilot. From the cold-read review: the main README attributed the parallel-batch ~10x multiplier to the blocking face's ordered default (the bench README says the default lands lower — the two now tell one story); pipeline_buffer_test.go gets its honest pre-existing comment back; the Makefile comment says the subjects focus 'covers' every parametrized suite rather than 'lists explicitly'. * fix(bench): feasible zero-copy load, exact b.N accounting; CI polish The zero-copy benchmark at 64KiB payloads drove 500 goroutines through a PoolSize-10 client — on a 4-vCPU CI runner that blows the default 3s write deadline, and the newly surfaced errors (previously swallowed into the ops/sec count) correctly failed the bench on two matrix cells. Large-payload sizes now run 100 goroutines with generous deadlines, so the benchmark measures the command path rather than write-deadline queueing; the B/op result is unchanged (~100x less at 64KiB). From Copilot's follow-up review: benchWindowed and the zero-copy driver partition b.N exactly across workers (the first b.N%goroutines take one extra op) instead of floor-and-min-1, so ns/op and allocs/op are per command at every -benchtime — the old split over-ran for small b.N and dropped the remainder for large; and the run-tests action invokes make -- "$MAKE_TARGET" so a target value starting with a dash cannot inject make options. Also backtick MaxConcurrentBatches/Unordered in the README prose the previous commit added — the spellchecker exempts code spans. * fix(autopipeline): full review round for the master merge Required fixes from the integration review: ReadStringInto's fast path gates on len instead of cap, so the trailing CRLF can never be written past the slice the caller handed over (a sub-slice of a packed buffer exposed trailing capacity belonging to the next segment — the pattern the zero-copy example itself teaches); cluster mapCmdsByNode is strictly all-or-nothing again for user pipelines, with fan-out-policy commands rejected at autopipeline submit through a new engine preflight hook (so a merged batch cannot be poisoned and the lone-command path agrees with batched dispatch); both pipeline read loops carry push-frame-mid-batch regression tests (the cluster one mutation-validated); and Submit() marks commands ready so result accessors before Wait() await instead of racing the dispatcher. Codex findings: a bad-conn push-drain error on the dedicated pipeline pool removes the connection instead of pooling a desynced reader; PipelineReadBufferSize gets the RESP3 minimum clamp on both construction paths; a CSC WithTimeout clone's Close delegates to the owner's Client.Close so cached autopipeliners stop and the closed flag flips. Engine: AutoPipelineOptions.MaxBatchBytes caps flushes by approximate payload volume and dispatch splits oversized drains into byte-bounded chunks (batches still complete once, after all chunks) — many large values flush as several bounded writes instead of one multi-megabyte burst that can stall a constrained link past its write deadline; dispatch goroutines recover panics into per-command errors instead of crashing the process (a plain client unwinds a hook panic into the caller; the engine's internal goroutines had nobody to recover); the outside-pipeline guard covers the full connection-hostile command set for raw Submit/Do; internal backstop timeouts surface ErrAutoPipelineTimeout rather than a misleading context.DeadlineExceeded; ErrHookShortCircuit and ErrSubmitBlockingFace are exported; the content-shard index is MinInt-safe. Docs and tests per the review: pending-command String() formatting documented as unsynchronized; AfterEach drains subjects before flushing in the module suites; the CI subjects job claims results parity rather than batch coverage; new tests cover RESP2 batched demux, blocking-face ctx cancel, dispatch-panic containment, byte-capped batching, Submit misuse under -race, push drains on both read loops, policy rejection on both pipeline kinds, and the never-past-len reader guarantee; assorted minors (govulncheck minor-series pin, apTestAddr in buffer tests, slotCache happens-before note, twin-path drift warning, Close worst-case and Validate caveats, EXPERIMENTAL markers on new exports). * test: flush through the subject before closing it in the module suites The previous commit ordered closeSubject() before rawClient.FlushDB — but the closer closes rawClient, so the flush silently failed on a closed client and state leaked across specs (SEARCH_INDEX_EXISTS on the next protocol iteration, red on every CI test cell). Flush through the SUBJECT instead, in the original order: on the async autopipeline face the flush is ordered after any queued writes and closeSubject's drain guarantees it executes, which is what the reorder was trying to achieve; on the plain and blocking paths it is identical to the old rawClient flush. * fix(autopipeline): deferred face returns immediately for diverted commands; timeseries AfterEach Typed blocking and connection-hostile commands (BLPOP, SHUTDOWN, ...) were executed inline on the deferred face, stalling the submitter for the full block duration — Do already honored the deferred contract via its background-goroutine escape hatch. The diverted submit path now routes through the same runOutsidePipeline, keeping each face's call shape: the blocking face runs synchronously, the deferred face returns immediately with a ready batch and the result accessors block (review finding by cursor on #3942). A regression test pins submission returning immediately while BLPOP still blocks server-side for its full window. Also: the timeseries suite gets the same subject-flush AfterEach the previous commit applied to json/probabilistic/vectorset — flushing via rawClient before closing the subject let a queued async write land after the flush. * test: count solo dispatches in the MaxBatchBytes test The counting hook only watched the pipeline hook, but a drain cycle that catches exactly one command takes the engine's lone-command fast path (ProcessHook, no pipeline) — on a loaded runner one of the ten commands split off that way and the total came up short. Record solo dispatches as size-1 entries, filter to the test's own SET commands so a fresh-conn handshake pipeline cannot pollute the counts, and allow the uncapped control one straggler split. * test: widen the sentinel resolution budget for loaded runners The regression test demanded the full sentinel resolution chain inside a 500ms context — routinely missed on loaded CI runners over docker networking. Three seconds keeps the signal (the retry-storm exhaustion it pins burned the whole context regardless of size) without the flake. * fix: legacy DisableIndentity mapping in cluster node options; align checkout action version The cluster node clientOptions mapped the deprecated DisableIndentity field from the NEW DisableIdentity option, silently ignoring a user-supplied legacy value — Ring and the failover client already copy the legacy field. Also bump the run-tests composite action's internal checkout to v7, matching the workflow-level version. Both pre-existing upstream, folded in on the way to master. * fix(autopipeline): delegate cluster-wide commands, abort chunks after failed prefix, share the closed flag across clones Three review findings (codex on #3942) plus a test seam: AutoPipeliner now delegates ClusterClient's cluster-wide method overrides (DBSize, ScriptLoad, ScriptFlush, ScriptExists) to the underlying client — the embedded cmdable shadowed them and sent each to a single random node. Chunked dispatch (MaxBatchBytes) stops dispatching remaining chunks once an earlier chunk dies on a transport-class failure, failing them with the same error: the unchunked path fails or retries the batch as a unit, so in the ordered stream later commands must not overtake a failed prefix. Per-command redis errors are normal chunk outcomes and do not abort. The pool-closed flag (baseClient.apClosed) becomes a shared *atomic.Bool copied by WithTimeout/clone: closing a clone closes the SHARED pools, so the parent's AutoPipeline getters now see ErrClosed instead of building a fresh flusher against dead pools. ClusterClient passes nil — its per-node clients own their pools. The maintnotifications Manager gets a shutdownTimeout field (default 10s, behavior unchanged) so the Close put-back path — a failed shutdown returns the pool hook and reopens the manager for a retry — is pinned by a test. Tests: TestAPClusterWideCommandsDelegate, TestDispatchChunkedAbortsAfterFailedPrefix (unit-level: through the public API the byte cap wakes the flusher mid-submission, so one logical batch may split into several drains and a single-dispatch assertion races), TestWithTimeoutCloneCloseBlocksParentResurrection, TestCloseRetriesAdditionalHookAfterShutdownError. * fix: String() awaits deferred commands; share pooled-conn helpers between main and pipeline pools Public String methods on commands read their result fields before awaiting, so formatting an in-flight deferred-autopipeline command from another goroutine raced with reply processing. Every String method now calls await() first — the same accessor contract as Err/Val/Result. The dispatch-gid guard in await keeps a hook that formats its own batch on the dispatch goroutine safe (it sees the not-yet-executed view instead of deadlocking), and cmdString remains the non-waiting internal formatter on rawErr. The hook-peek regression tests now format commands from inside hooks as well (review finding by ofek and cursor on #3942). withPipelineConn duplicated getConn/releaseConn by hand and had drifted again: the dedicated pipeline pool missed the client-side-cache post-read probe on release and the connection-create-time metric. Conn preparation and release now go through pool-parameterized helpers shared with the main path (initPooledConn, releaseConnToPool), closing the drift class. This also fixes a main-path slot leak — a conn that cannot be re-acquired after its handshake is now Removed from the pool instead of abandoned. The reported Acquire hang on Go 1.23+ timer channels is refuted empirically and pinned by regression hammers: under synchronous timer channels Stop returns TRUE for an expired-but-undelivered fire (the delivery is aborted), returning false only once the value was actually received — and the sole receiver is Acquire's own select, so a taken token implies an unreceived fire and the drain branch is unreachable. Under GODEBUG=asynctimerchan=1 (a consumer main module on go < 1.23) the fired value sits buffered and the drain consumes it before pooled Reset-reuse. The hammers race Release against timeout expiry under both semantics. * fix(autopipeline): register cluster node executors on batches; harden pooled semaphore timers A node-level hook reading a command result before next() — exactly what redisotel's tracing does on cluster node clients via OnNewNode plus rediscmd.CmdsString's Err() peek — permanently wedged the deferred cluster face: the read blocked on a batch that completes only after the node call returns, and the single dispatch-gid guard cannot vouch for the cluster fan-out goroutines. Batches now carry a registry of node executor goroutines, entered and left around each node call's hook chain, and the accessor guards (await, Wait, WaitContext) treat registered goroutines like the dispatcher: result reads get the not-yet-executed view instead of self-deadlocking. Reproduced against a live cluster and pinned by TestClusterNodeHookResultPeekNoDeadlock. The pooled semaphore timers get a portable drain: it moves to a single putSemTimer at pool put-back using select-with-default, correct under both timer-channel semantics. Under go >= 1.23 synchronous channels Stop discards a fired-but-undelivered value, so the drain branch is unreachable — the default arm is a safety net so no semantics shift can ever turn it into a blocking receive. Under GODEBUG=asynctimerchan=1 (a consumer main module on go < 1.23) the buffered fire is consumed before Reset-reuse. Stale-fire hammers pin both semantics against spurious instant timeouts. * test: make the search RESP2/RESP3 equivalence suite hermetic The subtests drop only the index between runs, so earlier subtests' hashes were re-indexed by the next one (the aggregate subtest saw the search subtest's third document), and waitForIndexing polled FT.INFO a hundred times without sleeping — returning in microseconds on a partial count. On a slow runner the index was still churning between the RESP2 and RESP3 queries, which surfaced as a transient extra empty row in the RESP3 aggregate reply. Each subtest now starts from a flushed keyspace, and the wait requires the exact document count with ingestion settled (Indexing == 0), sleeping between tries. * test: wait for search ingestion in the home_json doctests The search examples write three documents and FT.SEARCH them back to back, but indexing is asynchronous — on a loaded runner the hash example's query raced ingestion and returned zero results. Both examples now wait for the exact document count with ingestion settled before querying. The wait lives in a separate non-example file and is called inside REMOVE_START/REMOVE_END markers, so the generated documentation snippets are unchanged. * test: assert module-suite teardown; scale semaphore hammers under -short The json, probabilistic, timeseries and vectorset suites ignored the results of their AfterEach FlushDB and closeSubject calls. On the async autopipeline subject the flush is enqueued and its failure only observable through the command's Err(), while AutoPipeliner.Close reports nil — so a failed teardown could silently leak keys into later specs. Both steps are asserted now, and awaiting the flush also pins it to execute at teardown time rather than inside the close drain. The semaphore timer hammers scale down under -short (200k -> 2k race iterations, 200 -> 20 stale-fire rounds), following the repo's existing short-mode convention for stress tests; full runs keep full coverage. * perf(autopipeline): restore the accessor guards' no-executor fast path The node-executor guard replaced the dispatch-gid short-circuit in the accessor guards (await, Wait, WaitContext) with a check that parsed the goroutine id and took the batch mutex unconditionally. Every blocking-face command and every pre-done future passes that guard once per wait, so concurrent waiters serialized on one batch's mutex with a stack parse each: the blocking face dropped from ~830k to ~138k ops/sec (6x) on the loopback throughput benchmark, while the windowed face was unaffected (its reads mostly land after batch completion). isExecutorGoroutine now answers the common case — dispatch gid unset and no registered node executors — from two atomic loads before any goroutine id parse or lock; the mutex is reachable only inside a cluster batch's node fan-out window. Blocking-face throughput measured back at the pre-regression band, and the node-hook deadlock regression test still passes: the guard still vouches for cluster node executors. * fix(autopipeline): honor shared pool closure at submit; stamp ready under the stripe lock; divert conn-state cluster commands Three review findings (codex on #3942): A cached AutoPipeliner kept accepting enqueues after a WithTimeout clone closed the SHARED pools — the shared closed flag guarded only the getters. It now threads into the pipeliner and every submit path (batched and diverted) checks it, so work is refused with ErrClosed at submit instead of failing dispatch by dispatch against dead pools. The deferred face published a command to its shard before setReady stamped the gating batch, so a fast flush could take a command whose readyBatch was still nil — the cluster node-executor registration would skip that batch, and a node hook reading the command mid-dispatch could block on a batch its own call chain completes. enqueue now stamps ready under the stripe lock, before the command is visible to any drain; the ordering is guaranteed by the lock both sides take. readonly, readwrite and asking join the outside-pipeline divert set: they mutate connection-scoped cluster state (replica reads, a pending redirect) that must not leak onto shared pipeline connections — and ReadOnly/ReadWrite are reachable from the typed surface, not just Do. Tests: TestWithTimeoutCloneCloseStopsCachedAutoPipeliner, TestEnqueueStampsReadyUnderStripeLock, TestConnStateClusterCommandsRunOutsidePipeline. Throughput bands re-measured unchanged after the enqueue-path change. * fix(autopipeline): bound Close on tracked diverted dispatches; divert raw blocking commands; delegate HIMPORT Six review findings (codex, cxljs and ofek on #3942), all reproduced in the code before fixing. Close and diverted commands (one coupled change). Commands that run outside a batch — blocking and connection-hostile ones — executed on untracked goroutines, so Close could return while such a command still held a pooled connection. They are tracked now. Tracking alone would trade that leak for a worse bug: an accepted command is deliberately never cancelled by Close (the flush contract the batched path documents), so a zero-timeout blocking command, or a stalled read with ReadTimeout disabled, would hang Close forever. The wait is therefore bounded by the same backstop the permit wait uses, and on expiry Close returns an error naming what is still in flight rather than blocking its caller. Raw blocking commands. Typed helpers mark themselves with a per-command read timeout, which submit already diverts on; a raw Cmder built by hand carries no such marker, so BLPOP and friends could be queued onto a shared pipeline connection and hold it for the block duration. They are recognized by name now, with XREAD/XREADGROUP classified by their BLOCK argument so the common non-blocking form stays batched. Cluster-wide overrides. The HIMPORT admin trio joins the delegation set: like DBSize and the SCRIPT admin commands, ClusterClient fans them out to every master and updates the shared fieldset registry, so running them on a single routed node lets a later HImportSet for a key on another master fail. A reflection test now walks ClusterClient's method set and fails when an override that shadows the generic cmdable implementation has no delegate — this class has been found by review twice. Maintenance notifications. A delayed idle handoff for a connection from a dedicated pipeline pool queued on the primary pool hook, whose request carries the primary pool; a failed handoff then closed the connection without freeing its slot in the pipeline pool. Handoffs now route to the hook that owns the connection's pool, matched by the pool name already stamped on every connection. Close's bound is its own named constant rather than a borrowed one, with the reasoning recorded next to it: the bound is only reachable when a dispatch cannot end by itself, so a tighter value would not speed up healthy shutdowns — it would make Close report failure while a legitimate slow drain (a large final batch, a relaxed maintenance window during failover) is still finishing. Docs. The slot-to-shard mapping is described as what it is — a batch-depth heuristic. With non-contiguous slot ownership a shard's batch can still span nodes; correctness is unaffected because the node router resolves each command's own slot, and a key always lands on one shard so its relative order holds. Also lands the truth battery: thirteen end-to-end tests that check order, reply demultiplexing, exactly-once execution, at-most-once under retries, chunked-dispatch integrity, pipeline-pool isolation, cancellation honesty and the Close drain against an INDEPENDENT ground truth — a client-side model value, a server-side structure recording the real execution order, or a differential run of the same randomized workload through a plain client. * fix(autopipeline): close the divert registration race, honor shared close everywhere, order limiter reporting Five review findings on the previous commit (cursor Bugbot and codex on #3942), each reproduced in the code first. Shared close. Do, DoRaw and DoRawWriteTo guarded on this pipeliner's own closed flag rather than the combined check, so once a WithTimeout clone had closed the shared pool set they still accepted commands and dispatched against pools that were already gone. IsClosed had the same gap while its documentation promised that the owning client's close counts. Every guard that gates accepting new work now goes through the combined check, with the invariant recorded beside it: only Close's own compare-and-swap reads the instance flag directly, because it claims the shutdown rather than testing for it. Divert registration race. A diverted command could pass its closed check, Close could then observe an empty wait group and return, and only afterwards would the goroutine register — leaving an accepted command holding a pooled connection past Close, and racing Add against Wait. Registration now happens under a mutex with a closed re-check, and Close passes through that mutex once after marking itself closed, so a registration either completes before the wait or is rejected. The gate precedes publishing the command's batch: publishing first and then rejecting would leave the command gated on a batch nobody closes. Blocking detection. The BLOCK token is matched the way the encoder renders it — a raw Cmder may carry RESP keywords as []byte or *string, and a string-only type assertion let those XREAD forms be batched onto a shared connection. Cluster preflight. Diversion is decided before the preflight runs. The preflight keeps fan-out-policy commands out of shared pipelines, but a diverted command never rides one: it goes through Client/ClusterClient Process, where the cluster-wide aggregation works. Running the preflight first rejected typed WAIT/WAITAOF on a cluster whenever command policies were enabled. Limiter ordering. The dedicated pipeline-pool path registered the limiter report and the connection release as two defers, which run LIFO — the connection went back to the pool before the result was reported, so a circuit breaker could admit another pipelined operation before observing the failure. The main pool reports first; both paths now share one deferred exit that preserves that order and still reports on the acquire and handshake failure paths. * fix(autopipeline): bound the whole close drain; gate raw Do and telemetry on the divert contract Four review findings from codex plus two from Copilot, each reproduced in the code first. Raw Do on the blocking face. Do, DoRaw and DoRawWriteTo carried their own inline blocking-face branch that called Process directly, bypassing runOutsidePipeline and therefore the divert registration gate added in the previous commit. A raw blocking Do was untracked, so Close could observe no diverted work and tear down shared resources while it still held a pooled connection. Both faces now go through the one path that registers. Close's bound covered only the last wait. A dispatch that nothing can end sits just as easily in a flusher the wait group is waiting for, or in the shutdown sweep's own dispatch, as in the final wait — so Close hung before ever reaching the bound it documents. The whole drain tail (flusher exit, final shard sweep, batch and diverted waits) now runs under a single bound, and on expiry the error names the precise stage still outstanding, because a wedged flusher and a wedged blocking command call for different operator responses. Telemetry no longer changes the async call shape. The metric emissions after Publish, SPublish and XReadGroup read the command's outcome immediately, which AWAITS on the deferred face: with a recorder installed, an autopipelined publish blocked until Redis replied, and an XReadGroup with the default Block: 0 could block indefinitely in what should be a submit. They now consult a non-blocking readiness check and skip the metric for a submission that has not executed; emitting these from the execution path is a follow-up in the OTel wiring rather than something a command wrapper can do. Two Copilot findings: the ConnPool.Name doc block added last round swallowed Size()'s Godoc, leaving an exported method undocumented — restored; and the pipeline-buffer tests left fixed keys behind, one of which (test_key) is also used by another test file in the package, which is exactly the order-dependence that produced the search-suite flake earlier in this PR. * fix(autopipeline): stop Config leaking internal sharding; keep Eval non-blocking on the deferred face Two review findings from codex plus one from Copilot. Config leaked an internal invariant. The effective config returned to callers carried contentSharded, the bit cluster wiring sets to record that shards are slot-routed and therefore cannot reorder same-key commands. Copying that config into a standalone deferred autopipeliner silenced the check that requires Unordered for NumShards>1 — and round-robin shards genuinely do flush concurrently and break submit order. Internal-only fields are now cleared before the copy is handed out. Eval was synchronous on the deferred face. Normalizing NOSCRIPT to ErrNoScript read the command's error immediately after submitting, which awaits execution, so the whole Eval family lost the deferred call shape — a caller submitting a window of scripts, or running one slow script, blocked in the submit. The normalization now runs only when the result is already readable, and Script.Run/RunRO accept the server's raw NOSCRIPT error so the EVAL fallback keeps working when it was skipped. The synchronous and blocking faces are unchanged; on the deferred face the raw error is byte-identical to ErrNoScript, so errors.Is and MatchError assertions still hold. Copilot: the pipeline-buffer key cleanup added in the previous commit ran before the reachability check, issuing a DEL against a server that might not be there and turning a unit-only run into a failure instead of a skip. The four call sites skip first now, and the helper documents that ordering. * fix(autopipeline): keep a no-retry command's policy off its batchmates; complete the blocking set A command that forbids retries — today the zero-copy reads, whose reply decodes into a caller-owned buffer that a retry could not un-write — disables retries for the entire slice it is dispatched in. In a shared autopipeline batch that silently stripped retries from unrelated callers' ordinary commands, which would have been retried had they been flushed on their own. Mixed batches are now partitioned by retry policy and each group dispatched separately; uniform batches are returned untouched, so only the mixed case pays a second dispatch. Splitting rather than diverting the no-retry commands keeps zero-copy reads pipelined — measured after the change at 507k ops/sec and 497 B/op for a 4 KiB value, still ahead of the regular path. The raw-blocking allowlist was incomplete: blmovem, migrate, and the argument-driven ts.read BLOCK form could be queued onto a shared pipeline connection when submitted as raw Cmders, holding it for the block duration. The set is now derived from the typed helpers that call setReadTimeout rather than enumerated by hand, with that derivation recorded beside it and a test that pins every name and both arg-driven forms. Also rewords PipelineReadBufferSize's tuning guidance, which described the write side — sizing so a batch flushes without overflowing mid-write — and cross-references the option that guidance actually belongs to. codex's remaining finding, that ClusterClient.PoolStats does not fold in the per-node pipeline pools, is a monitoring gap rather than a correctness one and is left as a follow-up. * fix(otel): read the global recorder under its lock when registering pools RegisterPools and UnregisterPools read globalRecorder directly while SetGlobalRecorder writes it under recorderMu, so installing telemetry concurrently with creating or closing a client is a data race. A regression test reproduces it — two race reports under -race before the change, clean after — and both helpers now go through getRecorder. Also fixes the pipeline-buffer key cleanup added earlier in this PR: its post-test sweep ran in a t.Cleanup, which executes after the test function's defers, and every caller defers client.Close(). The DEL therefore landed on a closed client and silently did nothing, leaving exactly the keys the helper exists to remove. The sweep now uses its own short-lived client and reports a failure rather than swallowing it (both found by Copilot on #3942). * fix(autopipeline): isolate retry policy without reordering; honor successful hook short-circuits The no-retry isolation added in the previous commit split a mixed batch into two policy groups and dispatched the retryable one first, which REORDERED the stream: a zero-copy read submitted before a write to the same key executed after it, so the read observed the new value on a face that promises submit order. The batch is now sliced into contiguous runs and dispatched in order — each dispatched pipeline is still policy-uniform, so a no-retry command cannot strip retries from batchmates, and no command overtakes another. Uniform batches are still one dispatch with no allocation. Both new tests fail if the grouping is reintroduced, including the end-to-end one that reads the wrong value. A hook that returns nil WITHOUT calling next has short-circuited successfully: it served the batch itself and set the command values, which is exactly what a caching or mocking hook does and what plain Pipeline and Client hooks are permitted to do. The engine rewrote that success into an error, so such a hook worked on a pipeline but failed every autopipelined batch. All five sites now propagate only the hook's own error, and the sentinel this PR added for the synthesized case is retired. Cluster commands whose routing is not slot-derived — ReqSpecial, such as FT.CURSOR READ, which is sticky to the node holding the cursor — are now diverted instead of coalesced. A solo flush reached ClusterClient.process and its special routing, but inside a batch mapCmdsByNode routes by slot and can send the cursor read to the wrong shard, which only shows up under concurrent traffic. Diverting keeps them working on their own connection rather than rejecting them (all three found by codex on #3942). * fix(autopipeline): share the prefix-abort rule; report submit rejections from Process Breaking a batch into ordered pieces happens in two places now — the MaxBatchBytes chunker and the retry-policy runs — and each independently got the same rule wrong: once a piece dies on a transport-class failure, the remaining pieces must not be dispatched, because the unchunked path fails or retries the batch as a unit and later commands must never overtake a failed prefix. The runs loop dispatched them anyway. Both callers now go through one dispatchSequential helper that owns the rule, so a third variant cannot drift out of line; per-command redis errors remain normal outcomes that do not stop the sequence. The deferred face's Process discarded submit-time rejections. A closed pipeliner or a cluster preflight refusal sets the error on the command and queues nothing, so returning nil told the caller the command was accepted when it would never run — and code reaching the engine through UniversalClient.Process sees only that return value. Those rejections are returned now, identified by the shared completed batch rather than by probing for an error, so execution failures still stay off this path: the call must not wait. The type documentation still said String()/%v on a pending deferred command does not wait. It has waited since the accessor round, and a test pins exactly that, so the doc was steering users toward a race that no longer exists. It now describes the real contract, including the dispatch-goroutine exception, and points at Name()/Args() for logging a submission without waiting (all three found by codex on #3942).
300 lines
7.1 KiB
Go
300 lines
7.1 KiB
Go
package redis
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// TestAdaptiveDelayCalculation tests the adaptive delay calculation logic
|
|
func TestAdaptiveDelayCalculation(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
maxBatchSize int
|
|
maxDelay time.Duration
|
|
adaptive bool
|
|
queueLen int
|
|
expected time.Duration
|
|
}{
|
|
// Disabled adaptive delay
|
|
{
|
|
name: "adaptive disabled, returns fixed delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: false,
|
|
queueLen: 50,
|
|
expected: 100 * time.Microsecond,
|
|
},
|
|
{
|
|
name: "adaptive disabled, zero delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 0,
|
|
adaptive: false,
|
|
queueLen: 50,
|
|
expected: 0,
|
|
},
|
|
|
|
// Enabled adaptive delay - 75% threshold
|
|
{
|
|
name: "75% full - no delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 75,
|
|
expected: 0,
|
|
},
|
|
{
|
|
name: "76% full - no delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 76,
|
|
expected: 0,
|
|
},
|
|
{
|
|
name: "100% full - no delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 100,
|
|
expected: 0,
|
|
},
|
|
|
|
// Enabled adaptive delay - 50% threshold
|
|
{
|
|
name: "50% full - 25% delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 50,
|
|
expected: 25 * time.Microsecond,
|
|
},
|
|
{
|
|
name: "60% full - 25% delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 60,
|
|
expected: 25 * time.Microsecond,
|
|
},
|
|
{
|
|
name: "74% full - 25% delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 74,
|
|
expected: 25 * time.Microsecond,
|
|
},
|
|
|
|
// Enabled adaptive delay - 25% threshold
|
|
{
|
|
name: "25% full - 50% delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 25,
|
|
expected: 50 * time.Microsecond,
|
|
},
|
|
{
|
|
name: "30% full - 50% delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 30,
|
|
expected: 50 * time.Microsecond,
|
|
},
|
|
{
|
|
name: "49% full - 50% delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 49,
|
|
expected: 50 * time.Microsecond,
|
|
},
|
|
|
|
// Enabled adaptive delay - <25% threshold
|
|
{
|
|
name: "24% full - 100% delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 24,
|
|
expected: 100 * time.Microsecond,
|
|
},
|
|
{
|
|
name: "10% full - 100% delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 10,
|
|
expected: 100 * time.Microsecond,
|
|
},
|
|
{
|
|
name: "1% full - 100% delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 1,
|
|
expected: 100 * time.Microsecond,
|
|
},
|
|
|
|
// Edge cases
|
|
{
|
|
name: "empty queue - no delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 0,
|
|
expected: 0,
|
|
},
|
|
{
|
|
name: "zero max delay - no delay",
|
|
maxBatchSize: 100,
|
|
maxDelay: 0,
|
|
adaptive: true,
|
|
queueLen: 50,
|
|
expected: 0,
|
|
},
|
|
|
|
// Different batch sizes
|
|
{
|
|
name: "small batch - 75% full",
|
|
maxBatchSize: 10,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 8,
|
|
expected: 0,
|
|
},
|
|
{
|
|
name: "small batch - 50% full",
|
|
maxBatchSize: 10,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 5,
|
|
expected: 25 * time.Microsecond,
|
|
},
|
|
{
|
|
name: "large batch - 75% full",
|
|
maxBatchSize: 1000,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 750,
|
|
expected: 0,
|
|
},
|
|
{
|
|
name: "large batch - 50% full",
|
|
maxBatchSize: 1000,
|
|
maxDelay: 100 * time.Microsecond,
|
|
adaptive: true,
|
|
queueLen: 500,
|
|
expected: 25 * time.Microsecond,
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
// Create autopipeliner with test config
|
|
ap := &AutoPipeliner{
|
|
config: &AutoPipelineOptions{
|
|
MaxBatchSize: tt.maxBatchSize,
|
|
MaxFlushDelay: tt.maxDelay,
|
|
AdaptiveDelay: tt.adaptive,
|
|
},
|
|
}
|
|
sh := &apShard{ap: ap, stripes: make([]apStripe, 1)}
|
|
sh.stripes[0].queueLen.Store(int32(tt.queueLen))
|
|
ap.shards = []*apShard{sh}
|
|
|
|
// Calculate delay
|
|
result := ap.calculateDelay(tt.queueLen)
|
|
|
|
// Verify result
|
|
if result != tt.expected {
|
|
t.Errorf("calculateDelay() = %v, want %v", result, tt.expected)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestAdaptiveDelayIntegerArithmetic verifies integer arithmetic correctness
|
|
func TestAdaptiveDelayIntegerArithmetic(t *testing.T) {
|
|
maxBatch := 100
|
|
maxDelay := 100 * time.Microsecond
|
|
|
|
ap := &AutoPipeliner{
|
|
config: &AutoPipelineOptions{
|
|
MaxBatchSize: maxBatch,
|
|
MaxFlushDelay: maxDelay,
|
|
AdaptiveDelay: true,
|
|
},
|
|
}
|
|
sh := &apShard{ap: ap, stripes: make([]apStripe, 1)}
|
|
ap.shards = []*apShard{sh}
|
|
|
|
for queueLen := 0; queueLen <= maxBatch; queueLen++ {
|
|
sh.stripes[0].queueLen.Store(int32(queueLen))
|
|
delay := ap.calculateDelay(queueLen)
|
|
|
|
// Verify delay is one of the expected values
|
|
switch {
|
|
case queueLen == 0:
|
|
if delay != 0 {
|
|
t.Errorf("queueLen=%d: expected 0, got %v", queueLen, delay)
|
|
}
|
|
case queueLen*4 >= maxBatch*3: // ≥75%
|
|
if delay != 0 {
|
|
t.Errorf("queueLen=%d (≥75%%): expected 0, got %v", queueLen, delay)
|
|
}
|
|
case queueLen*2 >= maxBatch: // ≥50%
|
|
if delay != maxDelay/4 {
|
|
t.Errorf("queueLen=%d (≥50%%): expected %v, got %v", queueLen, maxDelay/4, delay)
|
|
}
|
|
case queueLen*4 >= maxBatch: // ≥25%
|
|
if delay != maxDelay/2 {
|
|
t.Errorf("queueLen=%d (≥25%%): expected %v, got %v", queueLen, maxDelay/2, delay)
|
|
}
|
|
default: // <25%
|
|
if delay != maxDelay {
|
|
t.Errorf("queueLen=%d (<25%%): expected %v, got %v", queueLen, maxDelay, delay)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// BenchmarkCalculateDelay benchmarks the delay calculation
|
|
func BenchmarkCalculateDelay(b *testing.B) {
|
|
ap := &AutoPipeliner{
|
|
config: &AutoPipelineOptions{
|
|
MaxBatchSize: 100,
|
|
MaxFlushDelay: 100 * time.Microsecond,
|
|
AdaptiveDelay: true,
|
|
},
|
|
}
|
|
sh := &apShard{ap: ap, stripes: make([]apStripe, 1)}
|
|
sh.stripes[0].queueLen.Store(50)
|
|
ap.shards = []*apShard{sh}
|
|
|
|
b.ResetTimer()
|
|
for i := 0; i < b.N; i++ {
|
|
_ = ap.calculateDelay(sh.Len())
|
|
}
|
|
}
|
|
|
|
// BenchmarkCalculateDelayDisabled benchmarks with adaptive delay disabled
|
|
func BenchmarkCalculateDelayDisabled(b *testing.B) {
|
|
ap := &AutoPipeliner{
|
|
config: &AutoPipelineOptions{
|
|
MaxBatchSize: 100,
|
|
MaxFlushDelay: 100 * time.Microsecond,
|
|
AdaptiveDelay: false,
|
|
},
|
|
}
|
|
sh := &apShard{ap: ap, stripes: make([]apStripe, 1)}
|
|
sh.stripes[0].queueLen.Store(50)
|
|
ap.shards = []*apShard{sh}
|
|
|
|
b.ResetTimer()
|
|
for i := 0; i < b.N; i++ {
|
|
_ = ap.calculateDelay(sh.Len())
|
|
}
|
|
}
|