From c587bdb80c9c55903bec969d6be32b7a5da21932 Mon Sep 17 00:00:00 2001 From: javi11 Date: Tue, 8 Sep 2026 16:38:36 +0200 Subject: [PATCH 1/2] perf(import): hedge straggling fast-fail STATs instead of riding the 2 s attempt ceiling The release probe STATs up to 64 sampled articles in one attempt whose deadline is capped at 2 s. When all but one or two answer in ~150 ms and the rest sit queued behind the concurrent first-segment warm-up on the same connections, the attempt ran to the full 2 s before the stragglers were retried, adding ~2 s to the import of every affected release. Once 90% of a sweep has reported, the remaining ids are re-issued on a second StatMany after a short grace derived from the observed median latency; the first answer per id wins and both sweeps stop as soon as every id is in. Attempt accounting, inconclusive/dead-release verdicts and cancellation are unchanged. --- internal/importer/validation/fast_fail.go | 2 +- .../importer/validation/fast_fail_hedge.go | 126 ++++++++++++++ .../validation/fast_fail_hedge_test.go | 164 ++++++++++++++++++ 3 files changed, 291 insertions(+), 1 deletion(-) create mode 100644 internal/importer/validation/fast_fail_hedge.go create mode 100644 internal/importer/validation/fast_fail_hedge_test.go diff --git a/internal/importer/validation/fast_fail.go b/internal/importer/validation/fast_fail.go index c45b909ed..18fa468a2 100644 --- a/internal/importer/validation/fast_fail.go +++ b/internal/importer/validation/fast_fail.go @@ -65,7 +65,7 @@ func statIDsWithBoundedRetries( reported := make(map[string]bool, len(remaining)) transient := make(map[string]error, len(remaining)) - for result := range client.StatMany(statCtx, remaining, nntppool.StatManyOptions{Concurrency: maxConnections}) { + for result := range hedgedStatMany(statCtx, client, remaining, maxConnections) { if _, wanted := seen[result.MessageID]; !wanted { continue } diff --git a/internal/importer/validation/fast_fail_hedge.go b/internal/importer/validation/fast_fail_hedge.go new file mode 100644 index 000000000..ab6301bad --- /dev/null +++ b/internal/importer/validation/fast_fail_hedge.go @@ -0,0 +1,126 @@ +package validation + +import ( + "context" + "slices" + "sync" + "time" + + "github.com/javi11/nntppool/v4" + "github.com/kipsilabs/altmount/internal/pool" +) + +const ( + // hedgeReportedFraction is the share of a sweep that must have answered + // before the remainder counts as straggling. + hedgeReportedFraction = 0.9 + // hedgeGraceLatencyFactor scales the observed median STAT latency into the + // grace a straggler gets before it is re-issued. + hedgeGraceLatencyFactor = 3 + hedgeMinGrace = 250 * time.Millisecond + hedgeMaxGrace = 750 * time.Millisecond +) + +// hedgedStatMany runs one StatMany sweep over ids and, once most of it has +// answered, re-issues the ids still outstanding on a second sweep so a STAT +// queued behind slow traffic on one connection does not hold the whole attempt +// to its deadline. Each id is reported at most once, whichever sweep answers +// first; both sweeps are cancelled as soon as every id has reported. +func hedgedStatMany(ctx context.Context, client pool.NntpClient, ids []string, concurrency int) <-chan nntppool.StatManyResult { + out := make(chan nntppool.StatManyResult, len(ids)) + go func() { + defer close(out) + sweepCtx, cancel := context.WithCancel(ctx) + defer cancel() + + start := time.Now() + var mu sync.Mutex + reported := make(map[string]struct{}, len(ids)) + latencies := make([]time.Duration, 0, len(ids)) + threshold := hedgeThreshold(len(ids)) + + primary := client.StatMany(sweepCtx, ids, nntppool.StatManyOptions{Concurrency: concurrency}) + var hedge <-chan nntppool.StatManyResult + var grace <-chan time.Time + + deliver := func(r nntppool.StatManyResult) { + mu.Lock() + if _, dup := reported[r.MessageID]; dup { + mu.Unlock() + return + } + reported[r.MessageID] = struct{}{} + latencies = append(latencies, time.Since(start)) + done := len(reported) + mu.Unlock() + + out <- r + if done == len(ids) { + cancel() + } else if hedge == nil && grace == nil && done >= threshold { + grace = time.After(hedgeGrace(latencies)) + } + } + + for primary != nil || hedge != nil { + select { + case r, ok := <-primary: + if !ok { + primary = nil + continue + } + deliver(r) + case r, ok := <-hedge: + if !ok { + hedge = nil + continue + } + deliver(r) + case <-grace: + grace = nil + mu.Lock() + stragglers := make([]string, 0, len(ids)-len(reported)) + for _, id := range ids { + if _, ok := reported[id]; !ok { + stragglers = append(stragglers, id) + } + } + mu.Unlock() + if len(stragglers) == 0 { + continue + } + hedge = client.StatMany(sweepCtx, stragglers, nntppool.StatManyOptions{ + Concurrency: len(stragglers), + Skip: func(id string) bool { + mu.Lock() + defer mu.Unlock() + _, done := reported[id] + return done + }, + }) + } + } + }() + return out +} + +// hedgeThreshold is the reported count at which the rest of an n-id sweep is +// considered straggling. It is never below 2 and never n itself, so a one- or +// two-id sweep is simply waited out. +func hedgeThreshold(n int) int { + t := int(float64(n)*hedgeReportedFraction + 0.999) + if t >= n { + return n + } + return max(t, 2) +} + +// hedgeGrace turns the latencies observed so far into how long a straggler is +// given before it is re-issued: a few medians, clamped so a very fast provider +// is not hedged on jitter and a slow one is not waited out to the deadline. +func hedgeGrace(latencies []time.Duration) time.Duration { + sorted := slices.Clone(latencies) + slices.Sort(sorted) + median := sorted[len(sorted)/2] + return min(max(median*hedgeGraceLatencyFactor, hedgeMinGrace), hedgeMaxGrace) +} diff --git a/internal/importer/validation/fast_fail_hedge_test.go b/internal/importer/validation/fast_fail_hedge_test.go new file mode 100644 index 000000000..c26d621e7 --- /dev/null +++ b/internal/importer/validation/fast_fail_hedge_test.go @@ -0,0 +1,164 @@ +package validation + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/javi11/nntppool/v4" +) + +// delayedStatClient answers every STAT with the configured outcome after a +// per-id delay that applies only to the first STAT of that id; a re-issued STAT +// answers immediately, the way a hedge landing on an idle connection does. +type delayedStatClient struct { + *scriptedStatClient + firstDelay map[string]time.Duration + sweeps int +} + +func newDelayedStatClient(outcomes map[string][]error, firstDelay map[string]time.Duration) *delayedStatClient { + return &delayedStatClient{ + scriptedStatClient: newScriptedStatClient(outcomes), + firstDelay: firstDelay, + } +} + +func (c *delayedStatClient) StatMany(ctx context.Context, ids []string, _ nntppool.StatManyOptions) <-chan nntppool.StatManyResult { + out := make(chan nntppool.StatManyResult, len(ids)) + c.mu.Lock() + c.sweeps++ + c.mu.Unlock() + var wg sync.WaitGroup + for _, id := range ids { + wg.Add(1) + go func(id string) { + defer wg.Done() + c.mu.Lock() + attempt := c.calls[id] + c.calls[id]++ + sequence := c.outcomes[id] + var err error + if len(sequence) > 0 { + err = sequence[min(attempt, len(sequence)-1)] + } + delay := time.Duration(0) + if attempt == 0 { + delay = c.firstDelay[id] + } + c.mu.Unlock() + + if delay > 0 { + timer := time.NewTimer(delay) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + } + result := nntppool.StatManyResult{MessageID: id, Err: err} + if err == nil { + result.Result = &nntppool.StatResult{MessageID: id} + } + select { + case out <- result: + case <-ctx.Done(): + } + }(id) + } + go func() { + wg.Wait() + close(out) + }() + return out +} + +func (c *delayedStatClient) sweepCount() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.sweeps +} + +func probeFile(count int) []FastFailFile { + return []FastFailFile{{Filename: "movie.mkv", Segments: makeTestSegments("seg", count)}} +} + +func TestFastFailReleaseProbeHedgesStragglerStat(t *testing.T) { + straggler := "seg-40" + client := newDelayedStatClient(nil, map[string]time.Duration{straggler: 5 * time.Second}) + + start := time.Now() + missing, err := FastFailReleaseProbe(context.Background(), probeFile(64), fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil) + elapsed := time.Since(start) + + if err != nil { + t.Fatalf("FastFailReleaseProbe error = %v, want nil", err) + } + if missing { + t.Fatal("missing = true, want false: every article exists") + } + if elapsed > 1500*time.Millisecond { + t.Fatalf("probe took %s, want the straggler hedged well inside the 2 s attempt ceiling", elapsed) + } + if got := client.callCount(straggler); got != 2 { + t.Fatalf("straggler STATs = %d, want 2 (original + hedge)", got) + } + if got := client.callCount("seg-0"); got != 1 { + t.Fatalf("fast id STATs = %d, want 1: only stragglers are hedged", got) + } +} + +func TestFastFailReleaseProbeDoesNotHedgeUniformlySlowSweep(t *testing.T) { + delays := make(map[string]time.Duration, 64) + for i := range 64 { + delays[fmt.Sprintf("seg-%d", i)] = 400 * time.Millisecond + } + client := newDelayedStatClient(nil, delays) + + missing, err := FastFailReleaseProbe(context.Background(), probeFile(64), fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil) + if err != nil { + t.Fatalf("FastFailReleaseProbe error = %v, want nil", err) + } + if missing { + t.Fatal("missing = true, want false") + } + if got := client.sweepCount(); got != 1 { + t.Fatalf("StatMany sweeps = %d, want 1: a uniformly slow provider has no stragglers to hedge", got) + } +} + +func TestFastFailReleaseProbeHedgedMissIsDefinitive(t *testing.T) { + straggler := "seg-40" + client := newDelayedStatClient( + map[string][]error{straggler: {nntppool.ErrArticleNotFound}}, + map[string]time.Duration{straggler: 5 * time.Second}, + ) + + start := time.Now() + missing, err := FastFailReleaseProbe(context.Background(), probeFile(64), fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil) + if err != nil { + t.Fatalf("FastFailReleaseProbe error = %v, want nil for definitive miss", err) + } + if !missing { + t.Fatal("missing = false, want true: the hedged STAT answered 430") + } + if elapsed := time.Since(start); elapsed > 1500*time.Millisecond { + t.Fatalf("probe took %s, want the hedged 430 to settle it early", elapsed) + } +} + +func TestFastFailReleaseProbeHedgeRespectsCancellation(t *testing.T) { + client := newDelayedStatClient(nil, map[string]time.Duration{"seg-40": 5 * time.Second}) + // Every hedge answers immediately, so the only way this probe finishes fast + // AND reports cancellation is if the straggler wait observed ctx. + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + _, err := FastFailReleaseProbe(ctx, probeFile(64), fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil) + if err == nil { + t.Fatal("FastFailReleaseProbe error = nil, want caller cancellation to surface") + } +} From 53261e92a7b6ad6bf413b918a28996d5387eb23d Mon Sep 17 00:00:00 2001 From: javi11 Date: Tue, 8 Sep 2026 17:16:53 +0200 Subject: [PATCH 2/2] perf(import): keep retrying fast-fail STATs while the sweep converges MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The probe gave up after exactly three 2 s attempts regardless of what the provider was doing. When STATs were slow but arriving (46 → 31 → 15 unanswered across the three attempts, right after a run of dead-release sweeps had the provider crawling through 430 lookups) a healthy release was refused as inconclusive. Retries now continue for as long as each attempt shrinks the unanswered set, within a 15 s wall-clock budget; an attempt at or past the third that makes no progress still ends the sweep, so a stuck provider costs what it did before. The retry backoff is clamped at 400 ms instead of doubling without bound. --- internal/importer/validation/fast_fail.go | 37 +++++--- .../validation/fast_fail_patience_test.go | 89 +++++++++++++++++++ 2 files changed, 115 insertions(+), 11 deletions(-) create mode 100644 internal/importer/validation/fast_fail_patience_test.go diff --git a/internal/importer/validation/fast_fail.go b/internal/importer/validation/fast_fail.go index 18fa468a2..ce7ece47a 100644 --- a/internal/importer/validation/fast_fail.go +++ b/internal/importer/validation/fast_fail.go @@ -17,10 +17,17 @@ import ( ) const ( + // fastFailStatMaxAttempts is the least number of attempts a sweep gets + // before a non-converging attempt ends it as inconclusive. fastFailStatMaxAttempts = 3 fastFailRetryBaseDelay = 100 * time.Millisecond + fastFailRetryMaxDelay = 400 * time.Millisecond ) +// fastFailStatBudget caps the wall-clock a sweep may spend across all its +// attempts. A var so tests can shorten it. +var fastFailStatBudget = 15 * time.Second + var ( // ErrFastFailInconclusive means bounded retries could not establish whether // one or more sampled articles exist. It is deliberately distinct from a @@ -33,9 +40,11 @@ func isDefinitiveFastFailMiss(err error) bool { return errors.Is(err, nntppool.ErrArticleNotFound) } -// statIDsWithBoundedRetries checks ids up to fastFailStatMaxAttempts times. -// Successful and definitively missing ids leave the retry set immediately; -// only operational errors and unreported ids are retried. The returned map +// statIDsWithBoundedRetries checks ids, retrying for as long as each attempt +// shrinks the unanswered set (at least fastFailStatMaxAttempts times, within +// fastFailStatBudget overall). Successful and definitively missing ids leave +// the retry set immediately; only operational errors and unreported ids are +// retried. The returned map // contains only definitive misses. When stopOnMissing is true the first such // miss ends the sweep, preserving the release probe's fast-fail behavior. func statIDsWithBoundedRetries( @@ -60,7 +69,8 @@ func statIDsWithBoundedRetries( missing = make(map[string]error) var lastErr error - for attempt := 1; attempt <= fastFailStatMaxAttempts && len(remaining) > 0; attempt++ { + sweepStart := time.Now() + for attempt := 1; len(remaining) > 0; attempt++ { statCtx, cancel := context.WithTimeout(ctx, pool.StatManyTimeout(len(remaining), maxConnections, timeout)) reported := make(map[string]bool, len(remaining)) transient := make(map[string]error, len(remaining)) @@ -120,22 +130,27 @@ func statIDsWithBoundedRetries( next = append(next, id) } } + converging := len(next) < len(remaining) remaining = next if len(remaining) == 0 { return missing, nil, nil } - if attempt == fastFailStatMaxAttempts || releaseLooksDead(len(missing), len(ids)-len(remaining)) { - // Bounded retries exhausted, or the definitive answers so far - // already condemn the release: a sweep dominated by 430s is a dead - // post whose remaining STATs are only queued behind more 430s - // (each one costs the provider a slow spool lookup), so waiting - // them out adds tens of seconds and changes nothing. + delay := min(fastFailRetryBaseDelay<<(attempt-1), fastFailRetryMaxDelay) + // A sweep that is still shrinking is a slow provider answering, not a + // dead one, so it is followed until the budget runs out. It stops early + // when an attempt past the minimum made no progress, or when the + // definitive answers so far already condemn the release: a sweep + // dominated by 430s is a dead post whose remaining STATs are only queued + // behind more 430s (each one costs the provider a slow spool lookup), so + // waiting them out adds tens of seconds and changes nothing. + stalled := attempt >= fastFailStatMaxAttempts && !converging + overBudget := time.Since(sweepStart)+delay >= fastFailStatBudget + if stalled || overBudget || releaseLooksDead(len(missing), len(ids)-len(remaining)) { return missing, remaining, fmt.Errorf("%w: %d segment(s) remained unverified after %d attempts: %w", ErrFastFailInconclusive, len(remaining), attempt, lastErr) } - delay := fastFailRetryBaseDelay << (attempt - 1) slog.WarnContext(ctx, "Retrying inconclusive fast-fail STATs", "attempt", attempt+1, "remaining", len(remaining), diff --git a/internal/importer/validation/fast_fail_patience_test.go b/internal/importer/validation/fast_fail_patience_test.go new file mode 100644 index 000000000..8250e88b7 --- /dev/null +++ b/internal/importer/validation/fast_fail_patience_test.go @@ -0,0 +1,89 @@ +package validation + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "github.com/javi11/nntppool/v4" +) + +// transientFor builds a STAT script that fails n times with a retryable error +// before answering "exists". +func transientFor(n int) []error { + seq := make([]error, 0, n+1) + for range n { + seq = append(seq, nntppool.ErrConnectionDied) + } + return append(seq, nil) +} + +// convergingScript scripts ids so that each retry clears one more tenth of the +// sweep: ids 0-9 answer on the first attempt, 10-19 need two, and so on. +func convergingScript(count, maxFailures int) map[string][]error { + outcomes := make(map[string][]error, count) + for i := range count { + outcomes[fmt.Sprintf("seg-%d", i)] = transientFor(min(i/10, maxFailures)) + } + return outcomes +} + +func TestFastFailReleaseProbeKeepsRetryingWhileSweepConverges(t *testing.T) { + // 40 ids, the last ten needing four attempts: every attempt reports more + // than the one before, so the probe should see it through rather than give + // up at the old hard cap of three. + client := newScriptedStatClient(convergingScript(40, 4)) + + missing, err := FastFailReleaseProbe(context.Background(), probeFile(40), fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil) + if err != nil { + t.Fatalf("FastFailReleaseProbe error = %v, want nil: the sweep was converging", err) + } + if missing { + t.Fatal("missing = true, want false") + } + if got := client.callCount("seg-39"); got != 4 { + t.Fatalf("slowest id STATs = %d, want 4", got) + } +} + +func TestFastFailReleaseProbeStopsOnceSweepStallsAfterMinimumAttempts(t *testing.T) { + outcomes := convergingScript(20, 1) + // ids 10-19 clear on attempt 2; these never answer. + for i := 20; i < 30; i++ { + outcomes[fmt.Sprintf("seg-%d", i)] = []error{nntppool.ErrConnectionDied} + } + client := newScriptedStatClient(outcomes) + + _, err := FastFailReleaseProbe(context.Background(), probeFile(30), fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil) + if !errors.Is(err, ErrFastFailInconclusive) { + t.Fatalf("FastFailReleaseProbe error = %v, want ErrFastFailInconclusive once progress stops", err) + } + if got := client.callCount("seg-25"); got != fastFailStatMaxAttempts { + t.Fatalf("stuck id STATs = %d, want %d: one non-converging attempt past the minimum ends the sweep", got, fastFailStatMaxAttempts) + } +} + +func TestFastFailReleaseProbeConvergenceIsBoundedByTotalBudget(t *testing.T) { + prev := fastFailStatBudget + fastFailStatBudget = 150 * time.Millisecond + t.Cleanup(func() { fastFailStatBudget = prev }) + + client := newScriptedStatClient(convergingScript(60, 5)) + + start := time.Now() + _, err := FastFailReleaseProbe(context.Background(), probeFile(60), fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil) + elapsed := time.Since(start) + if !errors.Is(err, ErrFastFailInconclusive) { + t.Fatalf("FastFailReleaseProbe error = %v, want ErrFastFailInconclusive when the budget runs out", err) + } + if elapsed > time.Second { + t.Fatalf("probe took %s, want it to stop near the %s budget", elapsed, fastFailStatBudget) + } + // Attempt 1 at t=0, 100 ms backoff, attempt 2 at ~100 ms; the next 200 ms + // backoff would overrun the 150 ms budget, so the sweep ends there. + if got := client.callCount("seg-59"); got != 2 { + t.Fatalf("slowest id STATs = %d, want 2: the budget must cut the sweep short", got) + } +}