perf(import): faster, steadier imports — hedged fast-fail STATs, warmed first articles, dead-post fast path

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,12 +69,13 @@ 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))
 
-		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
 			}
@@ -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),

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)
+}

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")
+	}
+}