From 2292a7dd057701b6cbb31a10d43e16ff68e8fb6d Mon Sep 17 00:00:00 2001 From: javi11 Date: Tue, 8 Sep 2026 18:10:47 +0200 Subject: [PATCH 01/11] fix(webdav): log a WARN when a streamed body ends on a read error http.ServeContent swallows the reader's error and stops writing, so a usenet fetch that gives up mid-stream reached the client as a truncated body with nothing in the server log to explain it. A bench playback died this way and the INFO log was empty. Wrap the file so the handler sees how the body ended and record the path, range, bytes served and error. --- internal/webdav/adapter.go | 10 +- internal/webdav/adapter_get_test.go | 139 ++++++++++++++++++++++++++++ internal/webdav/tracked_file.go | 25 +++++ 3 files changed, 173 insertions(+), 1 deletion(-) create mode 100644 internal/webdav/adapter_get_test.go create mode 100644 internal/webdav/tracked_file.go diff --git a/internal/webdav/adapter.go b/internal/webdav/adapter.go index f0a58c22e..e1cd113a8 100644 --- a/internal/webdav/adapter.go +++ b/internal/webdav/adapter.go @@ -148,7 +148,15 @@ func (h *webdavMethods) handleGet(w http.ResponseWriter, r *http.Request) { } defer f.Close() - http.ServeContent(w, r, fi.Name(), fi.ModTime(), f) + tracked := &readTracker{File: f} + http.ServeContent(w, r, fi.Name(), fi.ModTime(), tracked) + if tracked.err != nil { + slog.WarnContext(ctx, "WebDAV stream ended on read error", + "path", reqPath, + "range", r.Header.Get("Range"), + "bytes_served", tracked.bytesRead, + "error", tracked.err) + } } func (h *webdavMethods) handleDelete(w http.ResponseWriter, r *http.Request) { diff --git a/internal/webdav/adapter_get_test.go b/internal/webdav/adapter_get_test.go new file mode 100644 index 000000000..d3975cf33 --- /dev/null +++ b/internal/webdav/adapter_get_test.go @@ -0,0 +1,139 @@ +package webdav + +import ( + "bytes" + "context" + "errors" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "strings" + "testing" + "time" +) + +// failingFile serves size bytes of zeros but fails with failErr once failAt +// bytes have been read, the way a usenet reader does when an article fetch +// gives up mid-stream. +type failingFile struct { + size int64 + pos int64 + failAt int64 + failErr error +} + +func (f *failingFile) Read(p []byte) (int, error) { + if f.failErr != nil && f.pos >= f.failAt { + return 0, f.failErr + } + remaining := f.size - f.pos + if f.failErr != nil { + remaining = min(remaining, f.failAt-f.pos) + } + if remaining <= 0 { + return 0, io.EOF + } + n := int(min(int64(len(p)), remaining)) + clear(p[:n]) + f.pos += int64(n) + return n, nil +} + +func (f *failingFile) Seek(offset int64, whence int) (int64, error) { + switch whence { + case io.SeekStart: + f.pos = offset + case io.SeekCurrent: + f.pos += offset + case io.SeekEnd: + f.pos = f.size + offset + } + return f.pos, nil +} + +func (f *failingFile) Close() error { return nil } +func (f *failingFile) Write([]byte) (int, error) { return 0, errors.New("read-only") } +func (f *failingFile) Readdir(int) ([]os.FileInfo, error) { return nil, errors.New("not a dir") } +func (f *failingFile) Stat() (os.FileInfo, error) { return fileStat{size: f.size}, nil } + +type fileStat struct{ size int64 } + +func (s fileStat) Name() string { return "movie.mkv" } +func (s fileStat) Size() int64 { return s.size } +func (s fileStat) Mode() os.FileMode { return 0o644 } +func (s fileStat) ModTime() time.Time { return time.Unix(0, 0) } +func (s fileStat) IsDir() bool { return false } +func (s fileStat) Sys() any { return nil } + +type singleFileFS struct{ file *failingFile } + +func (s singleFileFS) Mkdir(context.Context, string, os.FileMode) error { return errors.New("ro") } +func (s singleFileFS) RemoveAll(context.Context, string) error { return errors.New("ro") } +func (s singleFileFS) Rename(context.Context, string, string) error { return errors.New("ro") } +func (s singleFileFS) Stat(context.Context, string) (os.FileInfo, error) { + return fileStat{size: s.file.size}, nil +} +func (s singleFileFS) OpenFile(context.Context, string, int, os.FileMode) (File, error) { + return s.file, nil +} + +func captureLogs(t *testing.T) *bytes.Buffer { + t.Helper() + var buf bytes.Buffer + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))) + t.Cleanup(func() { slog.SetDefault(prev) }) + return &buf +} + +func serveGet(t *testing.T, file *failingFile, rangeHeader string) *httptest.ResponseRecorder { + t.Helper() + methods := &webdavMethods{fs: singleFileFS{file: file}, prefix: "/webdav/"} + req := httptest.NewRequest(http.MethodGet, "/webdav/bench/movie.mkv", nil) + if rangeHeader != "" { + req.Header.Set("Range", rangeHeader) + } + rec := httptest.NewRecorder() + methods.handleGet(rec, req) + return rec +} + +func TestHandleGetWarnsWhenStreamEndsOnReadError(t *testing.T) { + logs := captureLogs(t) + fetchErr := errors.New("nntp: all providers exhausted") + file := &failingFile{size: 4 << 20, failAt: 1 << 20, failErr: fetchErr} + + rec := serveGet(t, file, "bytes=0-") + + if rec.Code != http.StatusPartialContent { + t.Fatalf("status = %d, want 206", rec.Code) + } + if got := rec.Body.Len(); got != 1<<20 { + t.Fatalf("body bytes = %d, want the 1 MiB served before the failure", got) + } + out := logs.String() + if !strings.Contains(out, "level=WARN") || !strings.Contains(out, "stream ended on read error") { + t.Fatalf("want a WARN that the stream ended on a read error, got logs:\n%s", out) + } + for _, want := range []string{"path=bench/movie.mkv", "bytes_served=1048576", "all providers exhausted"} { + if !strings.Contains(out, want) { + t.Errorf("WARN is missing %q:\n%s", want, out) + } + } +} + +func TestHandleGetDoesNotWarnOnCleanRead(t *testing.T) { + logs := captureLogs(t) + file := &failingFile{size: 2 << 20} + + rec := serveGet(t, file, "bytes=0-") + + if rec.Code != http.StatusPartialContent || rec.Body.Len() != 2<<20 { + t.Fatalf("status = %d body = %d, want 206 with the whole file", rec.Code, rec.Body.Len()) + } + if strings.Contains(logs.String(), "level=WARN") { + t.Fatalf("unexpected WARN on a clean read:\n%s", logs.String()) + } +} diff --git a/internal/webdav/tracked_file.go b/internal/webdav/tracked_file.go new file mode 100644 index 000000000..dee80d734 --- /dev/null +++ b/internal/webdav/tracked_file.go @@ -0,0 +1,25 @@ +package webdav + +import ( + "errors" + "io" +) + +// readTracker wraps a File so the handler can learn how a body ended: +// http.ServeContent swallows the reader's error and just stops writing, which +// the client sees as a truncated body with nothing on the server side to +// explain it. +type readTracker struct { + File + bytesRead int64 + err error +} + +func (t *readTracker) Read(p []byte) (int, error) { + n, err := t.File.Read(p) + t.bytesRead += int64(n) + if err != nil && !errors.Is(err, io.EOF) && t.err == nil { + t.err = err + } + return n, err +} From b559a1a5c9432fece7e9436a1a29607bf9b51fb9 Mon Sep 17 00:00:00 2001 From: javi11 Date: Tue, 8 Sep 2026 18:20:14 +0200 Subject: [PATCH 02/11] perf(import): publish warmed first articles to the streaming segment store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Import already fetches the first article of every file (WarmFirstSegments) but kept only a 16 KiB head for the parser, so the cold open that follows an import fetched the same article from the provider again: ~55 ms cold TTFB against ~5 ms warm in the bench, and a flat cost on every click→byte. The parser now hands the whole decoded body to the streaming segment store — the in-memory tier that is on by default, plus the disk cache when enabled — so the first read of a freshly imported file is a cache hit. Cold TTFB on the bench drops to 3–10 ms. --- cmd/altmount/cmd/serve.go | 7 ++ internal/importer/parser/parser.go | 20 +++++ .../importer/parser/parser_warm_store_test.go | 84 +++++++++++++++++++ internal/importer/processor.go | 6 ++ internal/importer/service.go | 8 ++ 5 files changed, 125 insertions(+) create mode 100644 internal/importer/parser/parser_warm_store_test.go diff --git a/cmd/altmount/cmd/serve.go b/cmd/altmount/cmd/serve.go index 51348f3fe..8668bfda5 100644 --- a/cmd/altmount/cmd/serve.go +++ b/cmd/altmount/cmd/serve.go @@ -20,6 +20,7 @@ import ( "github.com/kipsilabs/altmount/internal/arrs/registrar" "github.com/kipsilabs/altmount/internal/config" "github.com/kipsilabs/altmount/internal/health" + "github.com/kipsilabs/altmount/internal/importer/parser" "github.com/kipsilabs/altmount/internal/metadata" "github.com/kipsilabs/altmount/internal/nzbfilesystem/segcache" "github.com/kipsilabs/altmount/internal/pool" @@ -142,6 +143,12 @@ func runServe(cmd *cobra.Command, args []string) error { // Keep the memory tier from pushing the live heap over the soft limit: // under GC pressure the governor shrinks it, then restores it when calm. go cacheSource.RunPressureGovernor(ctx) + importerService.SetSegmentStore(func() parser.SegmentStore { + if store := cacheSource.Store(); store != nil { + return store + } + return nil + }) // Background PAR2 repair: repairs missing articles and serves the patched // payloads on the read path's hole branch. diff --git a/internal/importer/parser/parser.go b/internal/importer/parser/parser.go index 3617be64d..ac7ad5df3 100644 --- a/internal/importer/parser/parser.go +++ b/internal/importer/parser/parser.go @@ -75,12 +75,27 @@ type FirstSegmentData struct { FirstArticleMissingID string } +// SegmentStore is where decoded articles are kept for the streaming readers. +// The parser only ever writes to it. +type SegmentStore interface { + Put(messageID string, data []byte) error +} + // Parser handles NZB file parsing type Parser struct { poolManager pool.Manager // Pool manager for dynamic pool access getConfig config.ConfigGetter // Returns current config for connection limits log *slog.Logger // Logger for debug/error messages heads *headCache // what earlier parses learned from the wire + // segmentStore resolves the streaming segment store at fetch time (its + // capacity and tiers follow config), or nil when caching is off. + segmentStore func() SegmentStore +} + +// SetSegmentStore lets first articles fetched at import be published to the +// streaming segment store, so the cold open that follows is a cache hit. +func (p *Parser) SetSegmentStore(resolve func() SegmentStore) { + p.segmentStore = resolve } // Use conc pool for parallel processing with proper error handling @@ -1039,6 +1054,11 @@ func (p *Parser) fetchBodyWithRetry(ctx context.Context, cp pool.NntpClient, seg p.poolManager.UpdateDownloadProgress("", int64(len(result.Bytes))) } p.heads.put(segmentID, articleHead{meta: result.YEnc, bytes: clipHead(result.Bytes)}) + if p.segmentStore != nil { + if store := p.segmentStore(); store != nil { + _ = store.Put(segmentID, result.Bytes) + } + } return result, nil } if stderrors.Is(fetchErr, nntppool.ErrArticleNotFound) { diff --git a/internal/importer/parser/parser_warm_store_test.go b/internal/importer/parser/parser_warm_store_test.go new file mode 100644 index 000000000..5ce5eef5e --- /dev/null +++ b/internal/importer/parser/parser_warm_store_test.go @@ -0,0 +1,84 @@ +package parser + +import ( + "bytes" + "context" + "sync" + "testing" + + "github.com/javi11/nntppool/v4" + "github.com/javi11/nzbparser" + "github.com/kipsilabs/altmount/internal/testsupport/fakepool" +) + +type recordingStore struct { + mu sync.Mutex + puts map[string][]byte +} + +func (s *recordingStore) Get(string) ([]byte, bool) { return nil, false } +func (s *recordingStore) Put(id string, data []byte) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.puts == nil { + s.puts = map[string][]byte{} + } + s.puts[id] = append([]byte(nil), data...) + return nil +} + +func (s *recordingStore) get(id string) ([]byte, bool) { + s.mu.Lock() + defer s.mu.Unlock() + d, ok := s.puts[id] + return d, ok +} + +// The first article of every file is fetched at import anyway; handing the +// whole decoded body to the streaming segment store means the cold open a +// moment later is served from memory instead of a second provider round trip. +func TestWarmFirstSegmentsPublishesWholeArticleToSegmentStore(t *testing.T) { + article := bytes.Repeat([]byte("A"), 700*1024) + fp := fakepool.New() + fp.SetBehavior("vid-0", fakepool.SegmentBehavior{ + Bytes: article, + YEnc: nntppool.YEncMeta{FileName: "Real.Movie.2024.mkv", FileSize: int64(len(article)) * 2, Part: 1, PartSize: int64(len(article))}, + }) + nzb := &nzbparser.Nzb{Files: nzbparser.NzbFiles{ + {Filename: "Real.Movie.2024.mkv", Segments: nzbparser.NzbSegments{ + {Bytes: 720000, Number: 1, ID: "vid-0"}, + {Bytes: 720000, Number: 2, ID: "vid-1"}, + }}, + }} + store := &recordingStore{} + p := NewParser(newFakeFullPoolManager(fp), stormConfigGetter(4)) + p.SetSegmentStore(func() SegmentStore { return store }) + + p.WarmFirstSegments(context.Background(), nzb.Files) + + got, ok := store.get("vid-0") + if !ok { + t.Fatal("warm-up did not put vid-0 into the segment store") + } + if !bytes.Equal(got, article) { + t.Fatalf("stored %d bytes, want the whole %d-byte article (not the clipped head)", len(got), len(article)) + } + if _, ok := store.get("vid-1"); ok { + t.Fatal("warm-up stored vid-1, which it never fetched") + } +} + +func TestWarmFirstSegmentsWithoutStoreStillWarmsHeads(t *testing.T) { + fp := fakepool.New() + fp.SetBehavior("vid-0", fakepool.SegmentBehavior{Bytes: []byte("x"), YEnc: nntppool.YEncMeta{FileSize: 1, PartSize: 1}}) + nzb := &nzbparser.Nzb{Files: nzbparser.NzbFiles{ + {Filename: "Real.Movie.2024.mkv", Segments: nzbparser.NzbSegments{{Bytes: 10, Number: 1, ID: "vid-0"}}}, + }} + p := NewParser(newFakeFullPoolManager(fp), stormConfigGetter(4)) + p.SetSegmentStore(func() SegmentStore { return nil }) + + p.WarmFirstSegments(context.Background(), nzb.Files) + if got := fp.PerMessageCalls("vid-0"); got != 1 { + t.Fatalf("warm-up fetched vid-0 %d times, want 1", got) + } +} diff --git a/internal/importer/processor.go b/internal/importer/processor.go index 1f305ff84..fb5cb37b9 100644 --- a/internal/importer/processor.go +++ b/internal/importer/processor.go @@ -135,6 +135,12 @@ func (proc *Processor) SetPatchIndex(idx validation.PatchIndex) { proc.patchIndex = idx } +// SetSegmentStore publishes first articles fetched at import to the streaming +// segment store, so the cold open right after an import is a cache hit. +func (proc *Processor) SetSegmentStore(resolve func() parser.SegmentStore) { + proc.parser.SetSegmentStore(resolve) +} + // queueNzbRepair queues an NZB-mode repair for a release that was deferred // before import, so the repair plans straight from the NZB (there is no file // metadata to plan from yet). diff --git a/internal/importer/service.go b/internal/importer/service.go index d07d1a501..7fa96433f 100644 --- a/internal/importer/service.go +++ b/internal/importer/service.go @@ -225,6 +225,14 @@ func (s *Service) SetPatchIndex(idx validation.PatchIndex) { } } +// SetSegmentStore wires the streaming segment store into the importer so +// articles fetched at import are already cached when playback starts. +func (s *Service) SetSegmentStore(resolve func() parser.SegmentStore) { + if s.processor != nil { + s.processor.SetSegmentStore(resolve) + } +} + // GetPostProcessor returns the post-processor coordinator func (s *Service) GetPostProcessor() *postprocessor.Coordinator { return s.postProcessor From 9631fb22f67f5f48470f831e0f4c9c0c49c28b69 Mon Sep 17 00:00:00 2001 From: javi11 Date: Tue, 8 Sep 2026 18:22:24 +0200 Subject: [PATCH 03/11] perf(import): warm the largest clean-named video's first article when a segment store is wired Clean-named videos skip the import-time first-segment fetch to save bandwidth, which also meant the one file a player opens first never reached the segment store and paid a provider round trip on its cold open (~85 ms vs 3-10 ms for everything else). Warm just the largest such file, only when a store can keep the article; the skip is unchanged for the rest and whenever caching is off. --- internal/importer/parser/parser.go | 28 ++++++++++- .../importer/parser/parser_warm_store_test.go | 49 +++++++++++++++++++ 2 files changed, 76 insertions(+), 1 deletion(-) diff --git a/internal/importer/parser/parser.go b/internal/importer/parser/parser.go index ac7ad5df3..99481143c 100644 --- a/internal/importer/parser/parser.go +++ b/internal/importer/parser/parser.go @@ -1005,9 +1005,10 @@ func (p *Parser) WarmFirstSegments(ctx context.Context, files []nzbparser.NzbFil } maxFetch := max(min(min(len(files), p.getConfig().TotalProviderConnections()), maxFetchGoroutines), 1) warm := concpool.New().WithMaxGoroutines(maxFetch).WithContext(ctx) + primary := p.primaryVideoToWarm(files) for i := range files { file := &files[i] - if len(file.Segments) == 0 || shouldSkipFirstSegmentFetch(file) { + if len(file.Segments) == 0 || (shouldSkipFirstSegmentFetch(file) && i != primary) { continue } id := file.Segments[0].ID @@ -1019,6 +1020,31 @@ func (p *Parser) WarmFirstSegments(ctx context.Context, files []nzbparser.NzbFil _ = warm.Wait() } +// primaryVideoToWarm is the index of the largest video file whose first +// segment the warm-up would otherwise skip, or -1. Clean-named videos skip the +// fetch to save bandwidth, but the largest one is what a player opens first: +// when a segment store can keep the article, warming just that file turns the +// cold open into a cache hit for one article's worth of bandwidth. +func (p *Parser) primaryVideoToWarm(files []nzbparser.NzbFile) int { + if p.segmentStore == nil || p.segmentStore() == nil { + return -1 + } + best := -1 + for i := range files { + file := &files[i] + if !shouldSkipFirstSegmentFetch(file) { + continue + } + if _, video := skipEligibleVideoExtensions[strings.ToLower(filepath.Ext(file.Filename))]; !video { + continue + } + if best < 0 || file.Bytes > files[best].Bytes { + best = i + } + } + return best +} + // fetchBodyWithRetry fetches one article, retrying transient failures. A // genuine article-not-found (430/423) is permanent and returned at once: // transient errors — connection exhaustion, timeouts, resets — must not be diff --git a/internal/importer/parser/parser_warm_store_test.go b/internal/importer/parser/parser_warm_store_test.go index 5ce5eef5e..23ad4dfce 100644 --- a/internal/importer/parser/parser_warm_store_test.go +++ b/internal/importer/parser/parser_warm_store_test.go @@ -82,3 +82,52 @@ func TestWarmFirstSegmentsWithoutStoreStillWarmsHeads(t *testing.T) { t.Fatalf("warm-up fetched vid-0 %d times, want 1", got) } } + +func cleanVideos() nzbparser.NzbFiles { + seg := func(prefix string) nzbparser.NzbSegments { + return nzbparser.NzbSegments{ + {Bytes: 720000, Number: 1, ID: prefix + "-0"}, + {Bytes: 720000, Number: 2, ID: prefix + "-1"}, + {Bytes: 720000, Number: 3, ID: prefix + "-2"}, + } + } + return nzbparser.NzbFiles{ + {Filename: "Show.S01E01.1080p.WEB-DL.mkv", Bytes: 4 << 30, Segments: seg("e1")}, + {Filename: "Show.S01E02.1080p.WEB-DL.mkv", Bytes: 6 << 30, Segments: seg("e2")}, + } +} + +// Clean-named videos skip their first-segment fetch to save bandwidth, but the +// largest one is what a player opens first: with a segment store to keep the +// article in, warming just that file turns the cold open into a cache hit. +func TestWarmFirstSegmentsWarmsLargestVideoWhenStoreIsWired(t *testing.T) { + fp := fakepool.New() + fp.SetDefaultBehavior(fakepool.SegmentBehavior{Bytes: []byte("v"), YEnc: nntppool.YEncMeta{FileSize: 1, PartSize: 1}}) + store := &recordingStore{} + p := NewParser(newFakeFullPoolManager(fp), stormConfigGetter(4)) + p.SetSegmentStore(func() SegmentStore { return store }) + + p.WarmFirstSegments(context.Background(), cleanVideos()) + + if got := fp.PerMessageCalls("e2-0"); got != 1 { + t.Fatalf("largest video first segment fetched %d times, want 1", got) + } + if _, ok := store.get("e2-0"); !ok { + t.Fatal("largest video first segment not put into the segment store") + } + if got := fp.PerMessageCalls("e1-0"); got != 0 { + t.Fatalf("smaller video first segment fetched %d times, want 0: only the largest is warmed", got) + } +} + +func TestWarmFirstSegmentsKeepsSkippingCleanVideosWithoutStore(t *testing.T) { + fp := fakepool.New() + fp.SetDefaultBehavior(fakepool.SegmentBehavior{Bytes: []byte("v"), YEnc: nntppool.YEncMeta{FileSize: 1, PartSize: 1}}) + p := NewParser(newFakeFullPoolManager(fp), stormConfigGetter(4)) + + p.WarmFirstSegments(context.Background(), cleanVideos()) + + if got := fp.PerMessageCalls("e1-0") + fp.PerMessageCalls("e2-0"); got != 0 { + t.Fatalf("clean-named videos fetched %d first segments without a store, want 0", got) + } +} From 793597c3ecceec1d2011611368e40784d533577b Mon Sep 17 00:00:00 2001 From: javi11 Date: Tue, 8 Sep 2026 18:44:39 +0200 Subject: [PATCH 04/11] fix(webdav): do not warn when a stream ends because the client hung up A client abort cancels the request context and the reader surfaces that cancellation as its read error; a bench pass logged 36 of those as stream failures. Only a body cut short while the client was still listening is a server-side failure. --- internal/webdav/adapter.go | 5 ++++- internal/webdav/adapter_get_test.go | 19 +++++++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/internal/webdav/adapter.go b/internal/webdav/adapter.go index e1cd113a8..688a95903 100644 --- a/internal/webdav/adapter.go +++ b/internal/webdav/adapter.go @@ -150,7 +150,10 @@ func (h *webdavMethods) handleGet(w http.ResponseWriter, r *http.Request) { tracked := &readTracker{File: f} http.ServeContent(w, r, fi.Name(), fi.ModTime(), tracked) - if tracked.err != nil { + // A client that hangs up cancels the request context and the reader + // surfaces that cancellation; only a body cut short while the client was + // still listening is a server-side failure worth a warning. + if tracked.err != nil && ctx.Err() == nil { slog.WarnContext(ctx, "WebDAV stream ended on read error", "path", reqPath, "range", r.Header.Get("Range"), diff --git a/internal/webdav/adapter_get_test.go b/internal/webdav/adapter_get_test.go index d3975cf33..8cb7161b0 100644 --- a/internal/webdav/adapter_get_test.go +++ b/internal/webdav/adapter_get_test.go @@ -137,3 +137,22 @@ func TestHandleGetDoesNotWarnOnCleanRead(t *testing.T) { t.Fatalf("unexpected WARN on a clean read:\n%s", logs.String()) } } + +// A client that hangs up mid-stream is not a server-side failure: the reader +// surfaces the request's own cancellation, and that must not be logged as a +// read error. +func TestHandleGetDoesNotWarnWhenClientCancels(t *testing.T) { + logs := captureLogs(t) + file := &failingFile{size: 4 << 20, failAt: 1 << 20, failErr: context.Canceled} + methods := &webdavMethods{fs: singleFileFS{file: file}, prefix: "/webdav/"} + ctx, cancel := context.WithCancel(context.Background()) + req := httptest.NewRequest(http.MethodGet, "/webdav/bench/movie.mkv", nil).WithContext(ctx) + req.Header.Set("Range", "bytes=0-") + rec := httptest.NewRecorder() + cancel() + methods.handleGet(rec, req) + + if strings.Contains(logs.String(), "level=WARN") { + t.Fatalf("client cancellation logged as a read error:\n%s", logs.String()) + } +} From 4297133dc7b976618caa471d43cf3ae122ae336e Mon Sep 17 00:00:00 2001 From: javi11 Date: Tue, 8 Sep 2026 18:44:39 +0200 Subject: [PATCH 05/11] perf(import): hedge fast-fail stragglers from 75% reported, on the priority lane MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A bench pass showed 9 of 64 probe STATs queued behind other traffic while the other 55 answered in ~150 ms — under the 90% threshold, so no hedge fired and the import rode the 2 s attempt ceiling again. Arm the hedge at 75% and send the re-issued STATs down the priority lane: on the normal lane they would only join the queue that made the originals straggle. A uniformly slow or dead release never reaches the threshold, so it is not hedged. --- .../importer/validation/fast_fail_hedge.go | 12 ++++- .../validation/fast_fail_hedge_test.go | 45 +++++++++++++++++++ 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/internal/importer/validation/fast_fail_hedge.go b/internal/importer/validation/fast_fail_hedge.go index ab6301bad..af8ee5abe 100644 --- a/internal/importer/validation/fast_fail_hedge.go +++ b/internal/importer/validation/fast_fail_hedge.go @@ -12,8 +12,11 @@ import ( const ( // hedgeReportedFraction is the share of a sweep that must have answered - // before the remainder counts as straggling. - hedgeReportedFraction = 0.9 + // before the remainder counts as straggling. On a busy pool up to a + // seventh of a 64-STAT probe has been seen queued behind other traffic + // while the rest answered in ~150 ms; a uniformly slow or dead release + // never gets this far. + hedgeReportedFraction = 0.75 // hedgeGraceLatencyFactor scales the observed median STAT latency into the // grace a straggler gets before it is re-issued. hedgeGraceLatencyFactor = 3 @@ -89,8 +92,13 @@ func hedgedStatMany(ctx context.Context, client pool.NntpClient, ids []string, c if len(stragglers) == 0 { continue } + // The stragglers are, by construction, queued behind other + // normal-lane traffic; a hedge on the same lane would join the + // queue. The priority lane lets an idle connection pick these + // few bodyless requests up ahead of it. hedge = client.StatMany(sweepCtx, stragglers, nntppool.StatManyOptions{ Concurrency: len(stragglers), + Priority: true, Skip: func(id string) bool { mu.Lock() defer mu.Unlock() diff --git a/internal/importer/validation/fast_fail_hedge_test.go b/internal/importer/validation/fast_fail_hedge_test.go index c26d621e7..c537861b6 100644 --- a/internal/importer/validation/fast_fail_hedge_test.go +++ b/internal/importer/validation/fast_fail_hedge_test.go @@ -162,3 +162,48 @@ func TestFastFailReleaseProbeHedgeRespectsCancellation(t *testing.T) { t.Fatal("FastFailReleaseProbe error = nil, want caller cancellation to surface") } } + +// optionsRecordingClient remembers the options of every StatMany sweep. +type optionsRecordingClient struct { + *delayedStatClient + mu sync.Mutex + opts []nntppool.StatManyOptions +} + +func (c *optionsRecordingClient) StatMany(ctx context.Context, ids []string, opts nntppool.StatManyOptions) <-chan nntppool.StatManyResult { + c.mu.Lock() + c.opts = append(c.opts, opts) + c.mu.Unlock() + return c.delayedStatClient.StatMany(ctx, ids, opts) +} + +// Nine of sixty-four STATs queued behind other traffic is the shape seen on a +// busy pool; that many stragglers must still be hedged, and the hedge must go +// down the priority lane or it just joins the same queue. +func TestFastFailReleaseProbeHedgesLargerStragglerTailOnPriorityLane(t *testing.T) { + delays := make(map[string]time.Duration, 9) + for i := 40; i < 49; i++ { + delays[fmt.Sprintf("seg-%d", i)] = 5 * time.Second + } + client := &optionsRecordingClient{delayedStatClient: newDelayedStatClient(nil, delays)} + + start := time.Now() + missing, err := FastFailReleaseProbe(context.Background(), probeFile(64), fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil) + if err != nil || missing { + t.Fatalf("FastFailReleaseProbe = (%v, %v), want (false, nil)", missing, err) + } + if elapsed := time.Since(start); elapsed > 1500*time.Millisecond { + t.Fatalf("probe took %s, want the nine stragglers hedged inside the 2 s ceiling", elapsed) + } + client.mu.Lock() + defer client.mu.Unlock() + if len(client.opts) != 2 { + t.Fatalf("StatMany sweeps = %d, want 2 (primary + hedge)", len(client.opts)) + } + if client.opts[0].Priority { + t.Fatal("primary sweep must stay on the normal lane") + } + if !client.opts[1].Priority { + t.Fatal("hedge sweep must use the priority lane") + } +} From c12dc7406a4344a39543a3dcbab4342c0229e0d8 Mon Sep 17 00:00:00 2001 From: javi11 Date: Tue, 8 Sep 2026 19:09:05 +0200 Subject: [PATCH 06/11] perf(import): hedge fast-fail stragglers on an arrival lull instead of a reported fraction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A debug import showed the probe's real straggler shape: 42 of 64 STATs answered in ~150 ms and 22 sat to the 2 s ceiling, then 19 of those answered instantly on retry and the last 3 were hedged and answered in 50-90 ms on the priority lane. The stragglers are STATs pipelined on slow or cold connections, not slow articles, and their number varies from one to a third of the probe — a 75% reported threshold missed the case that mattered. Once at least 8 answers are in, every answer restarts a lull timer sized from the observed median; when the sweep goes quiet with ids outstanding, all of them are re-issued on the priority lane. Both hedge decisions and answers are logged at debug level so the next unexplained 2 s import can be read from the log. --- .../importer/validation/fast_fail_hedge.go | 90 ++++++++++--------- .../validation/fast_fail_hedge_test.go | 24 +++++ 2 files changed, 73 insertions(+), 41 deletions(-) diff --git a/internal/importer/validation/fast_fail_hedge.go b/internal/importer/validation/fast_fail_hedge.go index af8ee5abe..234ebf280 100644 --- a/internal/importer/validation/fast_fail_hedge.go +++ b/internal/importer/validation/fast_fail_hedge.go @@ -2,6 +2,7 @@ package validation import ( "context" + "log/slog" "slices" "sync" "time" @@ -11,24 +12,26 @@ import ( ) const ( - // hedgeReportedFraction is the share of a sweep that must have answered - // before the remainder counts as straggling. On a busy pool up to a - // seventh of a 64-STAT probe has been seen queued behind other traffic - // while the rest answered in ~150 ms; a uniformly slow or dead release - // never gets this far. - hedgeReportedFraction = 0.75 + // hedgeMinReported is how many answers a sweep needs before a lull in + // arrivals means anything: below it there is no picture of what normal + // latency looks like, so nothing is hedged. + hedgeMinReported = 8 // hedgeGraceLatencyFactor scales the observed median STAT latency into the - // grace a straggler gets before it is re-issued. + // lull — time since the last answer — after which the outstanding ids are + // 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. +// hedgedStatMany runs one StatMany sweep over ids and, once answers have been +// arriving and then stop for a grace period, re-issues every id still +// outstanding on a second, priority-lane sweep. The stragglers measured were +// STATs pipelined on slow or cold connections while the rest of the sweep +// answered in ~150 ms — anywhere from one id to a third of the probe — and a +// re-issue picked up by an idle connection answers in tens of milliseconds. +// 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() { @@ -40,28 +43,40 @@ func hedgedStatMany(ctx context.Context, client pool.NntpClient, ids []string, c 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 + lull := time.NewTimer(time.Hour) + lull.Stop() + defer lull.Stop() - deliver := func(r nntppool.StatManyResult) { + deliver := func(r nntppool.StatManyResult, fromHedge bool) { mu.Lock() if _, dup := reported[r.MessageID]; dup { mu.Unlock() return } reported[r.MessageID] = struct{}{} - latencies = append(latencies, time.Since(start)) + latency := time.Since(start) + latencies = append(latencies, latency) done := len(reported) mu.Unlock() + if hedge != nil { + slog.DebugContext(ctx, "hedged fast-fail STAT answered", + "segment_id", r.MessageID, + "from_hedge", fromHedge, + "latency", latency, + "error", r.Err) + } out <- r - if done == len(ids) { + switch { + case done == len(ids): cancel() - } else if hedge == nil && grace == nil && done >= threshold { - grace = time.After(hedgeGrace(latencies)) + case hedge == nil && done >= hedgeMinReported: + // Every answer restarts the lull: hedging begins only when + // the sweep has gone quiet with ids still outstanding. + lull.Reset(hedgeGrace(latencies)) } } @@ -72,15 +87,14 @@ func hedgedStatMany(ctx context.Context, client pool.NntpClient, ids []string, c primary = nil continue } - deliver(r) + deliver(r, false) case r, ok := <-hedge: if !ok { hedge = nil continue } - deliver(r) - case <-grace: - grace = nil + deliver(r, true) + case <-lull.C: mu.Lock() stragglers := make([]string, 0, len(ids)-len(reported)) for _, id := range ids { @@ -92,10 +106,14 @@ func hedgedStatMany(ctx context.Context, client pool.NntpClient, ids []string, c if len(stragglers) == 0 { continue } - // The stragglers are, by construction, queued behind other - // normal-lane traffic; a hedge on the same lane would join the - // queue. The priority lane lets an idle connection pick these - // few bodyless requests up ahead of it. + slog.DebugContext(ctx, "hedging straggling fast-fail STATs", + "stragglers", len(stragglers), + "reported", len(ids)-len(stragglers), + "elapsed", time.Since(start)) + // The stragglers are queued behind other normal-lane traffic; + // a hedge on the same lane would join the queue. The priority + // lane lets an idle connection pick these bodyless requests up + // ahead of it. hedge = client.StatMany(sweepCtx, stragglers, nntppool.StatManyOptions{ Concurrency: len(stragglers), Priority: true, @@ -112,20 +130,10 @@ func hedgedStatMany(ctx context.Context, client pool.NntpClient, ids []string, c 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. +// hedgeGrace turns the latencies observed so far into how long the sweep may +// stay silent before the outstanding ids are 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) diff --git a/internal/importer/validation/fast_fail_hedge_test.go b/internal/importer/validation/fast_fail_hedge_test.go index c537861b6..071bfa45f 100644 --- a/internal/importer/validation/fast_fail_hedge_test.go +++ b/internal/importer/validation/fast_fail_hedge_test.go @@ -207,3 +207,27 @@ func TestFastFailReleaseProbeHedgesLargerStragglerTailOnPriorityLane(t *testing. t.Fatal("hedge sweep must use the priority lane") } } + +// A third of a probe stuck behind slow connections while the other two thirds +// answered in ~150 ms is what a cold pool looks like; a fixed fraction never +// catches it. Once answers stop arriving for a grace period, whatever is still +// outstanding is hedged, however many that is. +func TestFastFailReleaseProbeHedgesWhenArrivalsStall(t *testing.T) { + delays := make(map[string]time.Duration, 22) + for i := 40; i < 62; i++ { + delays[fmt.Sprintf("seg-%d", i)] = 5 * time.Second + } + client := newDelayedStatClient(nil, delays) + + start := time.Now() + missing, err := FastFailReleaseProbe(context.Background(), probeFile(64), fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil) + if err != nil || missing { + t.Fatalf("FastFailReleaseProbe = (%v, %v), want (false, nil)", missing, err) + } + if elapsed := time.Since(start); elapsed > 1500*time.Millisecond { + t.Fatalf("probe took %s, want the 22 stalled STATs hedged inside the 2 s ceiling", elapsed) + } + if got := client.sweepCount(); got != 2 { + t.Fatalf("StatMany sweeps = %d, want 2 (primary + one hedge for every outstanding id)", got) + } +} From 04a9e534d4d22378515295f040c3c0d95e51270c Mon Sep 17 00:00:00 2001 From: javi11 Date: Tue, 8 Sep 2026 19:56:27 +0200 Subject: [PATCH 07/11] perf(import): let archive analysis read warmed articles back; pre-warm the last 7z volume's tail MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The RAR and 7z header passes fetch through an import-scoped cache that started empty, so the first volume's head — fetched moments earlier by the warm-up — was fetched again ("pass=7z-header hits 1 misses 2"), and the 7z end header at the tail of the last volume was a second serial round trip. Each cost ~250 ms on the import's critical path for every 7z release. The import cache now reads through to the streaming segment store the warm-up publishes into, and the warm-up also fetches the last two articles of the highest-numbered .7z.NNN volume when a store is wired. Both are no-ops when caching is off. The archive processors take the store through an optional setter so their interfaces and test doubles are unchanged. --- cmd/altmount/cmd/serve.go | 8 +-- internal/importer/archive/rar/processor.go | 27 ++++++++-- .../importer/archive/sevenzip/processor.go | 26 ++++++++- internal/importer/filesystem/segment_cache.go | 23 +++++++- .../filesystem/segment_cache_fallback_test.go | 47 ++++++++++++++++ internal/importer/parser/parser.go | 36 +++++++++++++ .../importer/parser/parser_warm_store_test.go | 53 +++++++++++++++++++ internal/importer/processor.go | 22 +++++++- internal/importer/service.go | 3 +- 9 files changed, 229 insertions(+), 16 deletions(-) create mode 100644 internal/importer/filesystem/segment_cache_fallback_test.go diff --git a/cmd/altmount/cmd/serve.go b/cmd/altmount/cmd/serve.go index 8668bfda5..3df6237f5 100644 --- a/cmd/altmount/cmd/serve.go +++ b/cmd/altmount/cmd/serve.go @@ -20,7 +20,6 @@ import ( "github.com/kipsilabs/altmount/internal/arrs/registrar" "github.com/kipsilabs/altmount/internal/config" "github.com/kipsilabs/altmount/internal/health" - "github.com/kipsilabs/altmount/internal/importer/parser" "github.com/kipsilabs/altmount/internal/metadata" "github.com/kipsilabs/altmount/internal/nzbfilesystem/segcache" "github.com/kipsilabs/altmount/internal/pool" @@ -143,12 +142,7 @@ func runServe(cmd *cobra.Command, args []string) error { // Keep the memory tier from pushing the live heap over the soft limit: // under GC pressure the governor shrinks it, then restores it when calm. go cacheSource.RunPressureGovernor(ctx) - importerService.SetSegmentStore(func() parser.SegmentStore { - if store := cacheSource.Store(); store != nil { - return store - } - return nil - }) + importerService.SetSegmentStore(cacheSource.Store) // Background PAR2 repair: repairs missing articles and serves the patched // payloads on the read path's hole branch. diff --git a/internal/importer/archive/rar/processor.go b/internal/importer/archive/rar/processor.go index 0bd8e12e8..a68857d13 100644 --- a/internal/importer/archive/rar/processor.go +++ b/internal/importer/archive/rar/processor.go @@ -9,6 +9,7 @@ import ( "strings" "time" + "github.com/javi11/rardecode/v2" "github.com/kipsilabs/altmount/internal/config" "github.com/kipsilabs/altmount/internal/errors" "github.com/kipsilabs/altmount/internal/importer/archive" @@ -19,7 +20,6 @@ import ( "github.com/kipsilabs/altmount/internal/pool" "github.com/kipsilabs/altmount/internal/progress" "github.com/kipsilabs/altmount/internal/usenet" - "github.com/javi11/rardecode/v2" ) // rarProcessor handles RAR archive analysis and content extraction @@ -27,6 +27,27 @@ type rarProcessor struct { log *slog.Logger poolManager pool.Manager configGetter config.ConfigGetter + // segmentStore resolves the streaming segment store, whose warmed first + // articles the header pass reads through; nil when caching is off. + segmentStore func() usenet.SegmentStore +} + +// SetSegmentStore lets analysis passes read articles the import warm-up already +// fetched instead of paying a second provider round trip for them. +func (rh *rarProcessor) SetSegmentStore(resolve func() usenet.SegmentStore) { + rh.segmentStore = resolve +} + +// newSegmentCache is the import-scoped cache for one analysis pass, reading +// through to the streaming store when one is wired. +func (rh *rarProcessor) newSegmentCache() *filesystem.ImportSegmentCache { + c := filesystem.NewImportSegmentCache(0) + if rh.segmentStore != nil { + if store := rh.segmentStore(); store != nil { + c.WithFallback(store) + } + } + return c } // NewProcessor creates a new RAR processor @@ -123,7 +144,7 @@ func (rh *rarProcessor) AnalyzeRarContentFromNzb(ctx context.Context, rarFiles [ // Import-scoped segment cache: rardecode's parallel volume reads and repeated // header probing frequently revisit the same leading segments across volumes. // Bounded and released (by dropping the reference) when this analysis pass returns. - segStore := filesystem.NewImportSegmentCache(0) + segStore := rh.newSegmentCache() defer segStore.LogStats(ctx, rh.log, "rar-header") ufs := filesystem.NewUsenetFileSystem(ctx, rh.poolManager, normalizedFiles, headerAnalysisPrefetch, progressTracker, readTimeout, segStore) @@ -801,7 +822,7 @@ func (rh *rarProcessor) processNestedRarContent(ctx context.Context, innerRarCon // Header analysis only reads initial volume headers, so prefetch is capped at 1. headerAnalysisPrefetch := 1 // Import-scoped segment cache, private to this nested-RAR analysis pass. - segStore := filesystem.NewImportSegmentCache(0) + segStore := rh.newSegmentCache() defer segStore.LogStats(ctx, rh.log, "rar-nested") dfs := filesystem.NewDecryptingFileSystem(ctx, rh.poolManager, entries, headerAnalysisPrefetch, readTimeout, segStore) diff --git a/internal/importer/archive/sevenzip/processor.go b/internal/importer/archive/sevenzip/processor.go index 47f345629..eb7dee999 100644 --- a/internal/importer/archive/sevenzip/processor.go +++ b/internal/importer/archive/sevenzip/processor.go @@ -22,6 +22,7 @@ import ( metapb "github.com/kipsilabs/altmount/internal/metadata/proto" "github.com/kipsilabs/altmount/internal/pool" "github.com/kipsilabs/altmount/internal/progress" + "github.com/kipsilabs/altmount/internal/usenet" "github.com/javi11/rardecode/v2" "github.com/javi11/sevenzip" "golang.org/x/text/encoding/unicode" @@ -32,6 +33,27 @@ type sevenZipProcessor struct { log *slog.Logger poolManager pool.Manager configGetter config.ConfigGetter + // segmentStore resolves the streaming segment store, whose warmed head and + // tail articles the header pass reads through; nil when caching is off. + segmentStore func() usenet.SegmentStore +} + +// SetSegmentStore lets analysis passes read articles the import warm-up already +// fetched instead of paying a second provider round trip for them. +func (sz *sevenZipProcessor) SetSegmentStore(resolve func() usenet.SegmentStore) { + sz.segmentStore = resolve +} + +// newSegmentCache is the import-scoped cache for one analysis pass, reading +// through to the streaming store when one is wired. +func (sz *sevenZipProcessor) newSegmentCache() *filesystem.ImportSegmentCache { + c := filesystem.NewImportSegmentCache(0) + if sz.segmentStore != nil { + if store := sz.segmentStore(); store != nil { + c.WithFallback(store) + } + } + return c } // NewProcessor creates a new 7zip processor @@ -135,7 +157,7 @@ func (sz *sevenZipProcessor) AnalyzeSevenZipContentFromNzb(ctx context.Context, // (see UsenetFile.ReadAt) — without a shared cache, every central-directory // or header probe that revisits an already-fetched segment re-downloads it. // Bounded and released (by dropping the reference) when this pass returns. - segStore := filesystem.NewImportSegmentCache(0) + segStore := sz.newSegmentCache() defer segStore.LogStats(ctx, sz.log, "7z-header") ufs := filesystem.NewUsenetFileSystem(ctx, sz.poolManager, sortedFiles, headerAnalysisPrefetch, progressTracker, readTimeout, segStore) @@ -918,7 +940,7 @@ func (sz *sevenZipProcessor) processNestedRarContent(ctx context.Context, innerR // Header analysis only reads initial volume headers, so prefetch is capped at 1. headerAnalysisPrefetch := 1 // Import-scoped segment cache, private to this nested-RAR analysis pass. - segStore := filesystem.NewImportSegmentCache(0) + segStore := sz.newSegmentCache() defer segStore.LogStats(ctx, sz.log, "7z-nested") dfs := filesystem.NewDecryptingFileSystem(ctx, sz.poolManager, entries, headerAnalysisPrefetch, readTimeout, segStore) diff --git a/internal/importer/filesystem/segment_cache.go b/internal/importer/filesystem/segment_cache.go index 0267bb9e3..52a91a831 100644 --- a/internal/importer/filesystem/segment_cache.go +++ b/internal/importer/filesystem/segment_cache.go @@ -57,6 +57,9 @@ type ImportSegmentCache struct { misses int64 evictions int64 curBytes int64 + + // fallback, when set, is consulted on a local miss; see WithFallback. + fallback usenet.SegmentStore } type importSegmentCacheEntry struct { @@ -153,6 +156,13 @@ func (c *ImportSegmentCache) Get(messageID string) ([]byte, bool) { el, ok := c.items[messageID] if !ok { + if c.fallback != nil { + if data, found := c.fallback.Get(messageID); found { + c.hits++ + c.putLocked(messageID, data) + return data, true + } + } c.misses++ return nil, false } @@ -167,7 +177,11 @@ func (c *ImportSegmentCache) Get(messageID string) ([]byte, bool) { func (c *ImportSegmentCache) Put(messageID string, data []byte) error { c.mu.Lock() defer c.mu.Unlock() + c.putLocked(messageID, data) + return nil +} +func (c *ImportSegmentCache) putLocked(messageID string, data []byte) { if el, ok := c.items[messageID]; ok { entry := el.Value.(*importSegmentCacheEntry) c.curBytes -= int64(len(entry.data)) @@ -191,6 +205,13 @@ func (c *ImportSegmentCache) Put(messageID string, data []byte) error { delete(c.items, entry.id) c.curBytes -= int64(len(entry.data)) } +} - return nil +// WithFallback makes Get consult store when this cache misses, keeping what it +// finds locally. The streaming segment store holds the articles the import +// warm-up fetched (first segments, the last 7z volume's tail), which are +// exactly the ones an archive-analysis pass reads first. +func (c *ImportSegmentCache) WithFallback(store usenet.SegmentStore) *ImportSegmentCache { + c.fallback = store + return c } diff --git a/internal/importer/filesystem/segment_cache_fallback_test.go b/internal/importer/filesystem/segment_cache_fallback_test.go new file mode 100644 index 000000000..352c9a4e5 --- /dev/null +++ b/internal/importer/filesystem/segment_cache_fallback_test.go @@ -0,0 +1,47 @@ +package filesystem + +import ( + "bytes" + "testing" +) + +type mapStore struct { + data map[string][]byte + gets int +} + +func (m *mapStore) Get(id string) ([]byte, bool) { + m.gets++ + d, ok := m.data[id] + return d, ok +} +func (m *mapStore) Put(id string, data []byte) error { m.data[id] = data; return nil } + +// Articles the import warm-up already fetched live in the streaming segment +// store; an analysis pass that misses its own cache should look there before +// paying a provider round trip, and keep what it finds for its next probe. +func TestImportSegmentCacheReadsThroughFallback(t *testing.T) { + article := bytes.Repeat([]byte("h"), 4096) + fallback := &mapStore{data: map[string][]byte{"head-0": article}} + c := NewImportSegmentCache(0).WithFallback(fallback) + + got, ok := c.Get("head-0") + if !ok || !bytes.Equal(got, article) { + t.Fatalf("Get(head-0) = (%d bytes, %v), want the fallback's article", len(got), ok) + } + if _, ok := c.Get("head-0"); !ok { + t.Fatal("second Get(head-0) missed: fallback hits must be kept locally") + } + if fallback.gets != 1 { + t.Fatalf("fallback consulted %d times, want 1", fallback.gets) + } + if s := c.Stats(); s.Hits != 2 || s.Misses != 0 { + t.Fatalf("stats = hits %d misses %d, want 2/0: a fallback hit is a hit", s.Hits, s.Misses) + } + if _, ok := c.Get("absent"); ok { + t.Fatal("Get(absent) hit") + } + if s := c.Stats(); s.Misses != 1 { + t.Fatalf("misses = %d, want 1 after an id neither tier has", s.Misses) + } +} diff --git a/internal/importer/parser/parser.go b/internal/importer/parser/parser.go index 99481143c..1ef51ca6c 100644 --- a/internal/importer/parser/parser.go +++ b/internal/importer/parser/parser.go @@ -1017,9 +1017,45 @@ func (p *Parser) WarmFirstSegments(ctx context.Context, files []nzbparser.NzbFil return nil }) } + for _, id := range p.sevenZipTailToWarm(files) { + warm.Go(func(ctx context.Context) error { + _, _ = p.fetchBodyWithRetry(ctx, cp, id) + return nil + }) + } _ = warm.Wait() } +// sevenZipTailToWarm is the last two segment ids of the highest-numbered +// .7z.NNN volume, or nil. 7z analysis opens the archive by reading the first +// volume's head and then the end header at the very end of the last volume; +// with a segment store to keep the articles in, both become cache hits instead +// of two serial provider round trips on the import's critical path. +func (p *Parser) sevenZipTailToWarm(files []nzbparser.NzbFile) []string { + if p.segmentStore == nil || p.segmentStore() == nil { + return nil + } + lastVolume, lastIdx := -1, -1 + for i := range files { + m := sevenZipContinuationPattern.FindStringSubmatch(files[i].Filename) + if m == nil { + continue + } + if n, err := strconv.Atoi(m[1]); err == nil && n > lastVolume { + lastVolume, lastIdx = n, i + } + } + if lastIdx < 0 { + return nil + } + segs := files[lastIdx].Segments + var ids []string + for i := len(segs) - 1; i > 0 && i >= len(segs)-2; i-- { + ids = append(ids, segs[i].ID) + } + return ids +} + // primaryVideoToWarm is the index of the largest video file whose first // segment the warm-up would otherwise skip, or -1. Clean-named videos skip the // fetch to save bandwidth, but the largest one is what a player opens first: diff --git a/internal/importer/parser/parser_warm_store_test.go b/internal/importer/parser/parser_warm_store_test.go index 23ad4dfce..5f0f8f8a1 100644 --- a/internal/importer/parser/parser_warm_store_test.go +++ b/internal/importer/parser/parser_warm_store_test.go @@ -131,3 +131,56 @@ func TestWarmFirstSegmentsKeepsSkippingCleanVideosWithoutStore(t *testing.T) { t.Fatalf("clean-named videos fetched %d first segments without a store, want 0", got) } } + +func sevenZipVolumes() nzbparser.NzbFiles { + seg := func(prefix string) nzbparser.NzbSegments { + return nzbparser.NzbSegments{ + {Bytes: 720000, Number: 1, ID: prefix + "-0"}, + {Bytes: 720000, Number: 2, ID: prefix + "-1"}, + {Bytes: 720000, Number: 3, ID: prefix + "-2"}, + {Bytes: 120000, Number: 4, ID: prefix + "-3"}, + } + } + return nzbparser.NzbFiles{ + {Filename: "Movie.7z.001", Bytes: 2 << 30, Segments: seg("v1")}, + {Filename: "Movie.7z.002", Bytes: 2 << 30, Segments: seg("v2")}, + {Filename: "Movie.7z.003", Bytes: 1 << 30, Segments: seg("v3")}, + } +} + +// 7z analysis reads the first volume's head and the last volume's tail, each +// a cold provider round trip today. With a store to keep them in, warm-up +// fetches the last volume's last two articles alongside the heads. +func TestWarmFirstSegmentsWarmsLastSevenZipVolumeTail(t *testing.T) { + fp := fakepool.New() + fp.SetDefaultBehavior(fakepool.SegmentBehavior{Bytes: []byte("z"), YEnc: nntppool.YEncMeta{FileSize: 1, PartSize: 1}}) + store := &recordingStore{} + p := NewParser(newFakeFullPoolManager(fp), stormConfigGetter(4)) + p.SetSegmentStore(func() SegmentStore { return store }) + + p.WarmFirstSegments(context.Background(), sevenZipVolumes()) + + for _, id := range []string{"v3-3", "v3-2"} { + if got := fp.PerMessageCalls(id); got != 1 { + t.Errorf("tail article %s fetched %d times, want 1", id, got) + } + if _, ok := store.get(id); !ok { + t.Errorf("tail article %s not put into the segment store", id) + } + } + if got := fp.PerMessageCalls("v2-3") + fp.PerMessageCalls("v1-3") + fp.PerMessageCalls("v3-1"); got != 0 { + t.Fatalf("unrelated articles fetched %d times, want 0", got) + } +} + +func TestWarmFirstSegmentsSkipsSevenZipTailWithoutStore(t *testing.T) { + fp := fakepool.New() + fp.SetDefaultBehavior(fakepool.SegmentBehavior{Bytes: []byte("z"), YEnc: nntppool.YEncMeta{FileSize: 1, PartSize: 1}}) + p := NewParser(newFakeFullPoolManager(fp), stormConfigGetter(4)) + + p.WarmFirstSegments(context.Background(), sevenZipVolumes()) + + if got := fp.PerMessageCalls("v3-3") + fp.PerMessageCalls("v3-2"); got != 0 { + t.Fatalf("tail articles fetched %d times without a store, want 0", got) + } +} diff --git a/internal/importer/processor.go b/internal/importer/processor.go index fb5cb37b9..29cc7399a 100644 --- a/internal/importer/processor.go +++ b/internal/importer/processor.go @@ -33,6 +33,7 @@ import ( "github.com/kipsilabs/altmount/internal/nzbfile" "github.com/kipsilabs/altmount/internal/pool" "github.com/kipsilabs/altmount/internal/progress" + "github.com/kipsilabs/altmount/internal/usenet" ) const ( @@ -137,8 +138,25 @@ func (proc *Processor) SetPatchIndex(idx validation.PatchIndex) { // SetSegmentStore publishes first articles fetched at import to the streaming // segment store, so the cold open right after an import is a cache hit. -func (proc *Processor) SetSegmentStore(resolve func() parser.SegmentStore) { - proc.parser.SetSegmentStore(resolve) +func (proc *Processor) SetSegmentStore(resolve func() usenet.SegmentStore) { + proc.parser.SetSegmentStore(func() parser.SegmentStore { + if store := resolve(); store != nil { + return store + } + return nil + }) + // The archive analysis passes read the warmed articles back through their + // import-scoped caches. Optional so the processors' interfaces (and their + // test doubles) stay unchanged. + type storeAware interface { + SetSegmentStore(func() usenet.SegmentStore) + } + if p, ok := proc.rarProcessor.(storeAware); ok { + p.SetSegmentStore(resolve) + } + if p, ok := proc.sevenZipProcessor.(storeAware); ok { + p.SetSegmentStore(resolve) + } } // queueNzbRepair queues an NZB-mode repair for a release that was deferred diff --git a/internal/importer/service.go b/internal/importer/service.go index 7fa96433f..6bd8cca65 100644 --- a/internal/importer/service.go +++ b/internal/importer/service.go @@ -34,6 +34,7 @@ import ( "github.com/kipsilabs/altmount/internal/nzbfile" "github.com/kipsilabs/altmount/internal/pool" "github.com/kipsilabs/altmount/internal/progress" + "github.com/kipsilabs/altmount/internal/usenet" "github.com/kipsilabs/altmount/internal/sabnzbd" "github.com/kipsilabs/altmount/internal/utils" "github.com/kipsilabs/altmount/pkg/rclonecli" @@ -227,7 +228,7 @@ func (s *Service) SetPatchIndex(idx validation.PatchIndex) { // SetSegmentStore wires the streaming segment store into the importer so // articles fetched at import are already cached when playback starts. -func (s *Service) SetSegmentStore(resolve func() parser.SegmentStore) { +func (s *Service) SetSegmentStore(resolve func() usenet.SegmentStore) { if s.processor != nil { s.processor.SetSegmentStore(resolve) } From b35f7339758e418ff84f654c8ceed09adac34686 Mon Sep 17 00:00:00 2001 From: javi11 Date: Tue, 8 Sep 2026 20:25:39 +0200 Subject: [PATCH 08/11] perf(import): take the parse's remaining round trips off the critical path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After every first segment is in, the parse still made two provider round trips before anything could be written: the representative middle segment's yEnc header (release-wide part size) and the PAR2 index, read through a reader with no cache so its first segment — fetched by the warm-up seconds earlier — was fetched again. ~400 ms of a season pack's 0.9 s import. The warm-up now fetches the representative header alongside the probe, and the PAR2 index reader is given the segment store the warm-up publishes into. Fetches that exist only to fill the store (the largest video's first article, the 7z tail) no longer hold up the parse: they run detached from the warm-up wait, bounded by their own timeout. --- internal/importer/parser/par2/descriptor.go | 6 +- internal/importer/parser/parser.go | 72 ++++++++++-- .../importer/parser/parser_warm_store_test.go | 111 ++++++++++++++++-- 3 files changed, 173 insertions(+), 16 deletions(-) diff --git a/internal/importer/parser/par2/descriptor.go b/internal/importer/parser/par2/descriptor.go index db27077d5..ea538de40 100644 --- a/internal/importer/parser/par2/descriptor.go +++ b/internal/importer/parser/par2/descriptor.go @@ -51,6 +51,7 @@ func GetFileDescriptors( ctx context.Context, firstSegmentCache []*FirstSegmentData, poolManager pool.Manager, + store usenet.SegmentStore, ) (map[[16]byte]*FileDescriptor, error) { descriptors := make(map[[16]byte]*FileDescriptor) @@ -73,7 +74,7 @@ func GetFileDescriptors( if len(cachedData.File.Segments) > MaxIndexSegments { continue // Skip large recovery block files } - fileDescriptors, err := readFileDescriptors(ctx, cachedData.File, poolManager) + fileDescriptors, err := readFileDescriptors(ctx, cachedData.File, poolManager, store) if err != nil { slog.DebugContext(ctx, "Failed to read PAR2 file descriptors, skipping", "error", err, "segments", len(cachedData.File.Segments)) @@ -97,6 +98,7 @@ func readFileDescriptors( ctx context.Context, par2File *nzbparser.NzbFile, poolManager pool.Manager, + store usenet.SegmentStore, ) ([]FileDescriptor, error) { var descriptors []FileDescriptor @@ -120,7 +122,7 @@ func readFileDescriptors( // Create UsenetReader (provides retry, prefetch, and metrics for free) rg := usenet.GetSegmentsInRange(ctx, 0, totalSize-1, loader) - r, err := usenet.NewUsenetReader(ctx, poolManager.GetPool, rg, 5, poolManager, "", nil, + r, err := usenet.NewUsenetReader(ctx, poolManager.GetPool, rg, 5, poolManager, "", store, usenet.WithImportProfile(poolManager)) if err != nil { return descriptors, fmt.Errorf("failed to create usenet reader: %w", err) diff --git a/internal/importer/parser/parser.go b/internal/importer/parser/parser.go index 1ef51ca6c..60f402d97 100644 --- a/internal/importer/parser/parser.go +++ b/internal/importer/parser/parser.go @@ -31,6 +31,7 @@ import ( "github.com/kipsilabs/altmount/internal/metadata" metapb "github.com/kipsilabs/altmount/internal/metadata/proto" "github.com/kipsilabs/altmount/internal/pool" + "github.com/kipsilabs/altmount/internal/usenet" "github.com/kipsilabs/altmount/internal/progress" "github.com/kipsilabs/altmount/internal/slogutil" "github.com/javi11/nntppool/v4" @@ -75,12 +76,25 @@ type FirstSegmentData struct { FirstArticleMissingID string } -// SegmentStore is where decoded articles are kept for the streaming readers. -// The parser only ever writes to it. +// SegmentStore is where decoded articles are kept for the streaming readers: +// warm-up writes the articles it fetches, and the PAR2 index read serves from +// it. Mirrors usenet.SegmentStore. type SegmentStore interface { + Get(messageID string) ([]byte, bool) Put(messageID string, data []byte) error } +// resolvedStore is the current segment store, or nil when caching is off. +func (p *Parser) resolvedStore() usenet.SegmentStore { + if p.segmentStore == nil { + return nil + } + if store := p.segmentStore(); store != nil { + return store + } + return nil +} + // Parser handles NZB file parsing type Parser struct { poolManager pool.Manager // Pool manager for dynamic pool access @@ -256,7 +270,7 @@ func (p *Parser) ParseNzb(ctx context.Context, n *nzbparser.Nzb, nzbPath string, par2Descriptors = cached return nil } - par2Descriptors, par2Err = par2.GetFileDescriptors(gctx, par2Cache, p.poolManager) + par2Descriptors, par2Err = par2.GetFileDescriptors(gctx, par2Cache, p.poolManager, p.resolvedStore()) if par2Err == nil { p.heads.putDescriptors(key, par2Descriptors) } @@ -1005,10 +1019,9 @@ func (p *Parser) WarmFirstSegments(ctx context.Context, files []nzbparser.NzbFil } maxFetch := max(min(min(len(files), p.getConfig().TotalProviderConnections()), maxFetchGoroutines), 1) warm := concpool.New().WithMaxGoroutines(maxFetch).WithContext(ctx) - primary := p.primaryVideoToWarm(files) for i := range files { file := &files[i] - if len(file.Segments) == 0 || (shouldSkipFirstSegmentFetch(file) && i != primary) { + if len(file.Segments) == 0 || shouldSkipFirstSegmentFetch(file) { continue } id := file.Segments[0].ID @@ -1017,15 +1030,60 @@ func (p *Parser) WarmFirstSegments(ctx context.Context, files []nzbparser.NzbFil return nil }) } - for _, id := range p.sevenZipTailToWarm(files) { + // The parse fetches one middle segment's yEnc header for the release-wide + // part size, after every first segment is in; fetched here it overlaps the + // probe and the parse finds it in the head cache. + if seg, groups, ok := representativeMiddleSegment(files); ok { warm.Go(func(ctx context.Context) error { - _, _ = p.fetchBodyWithRetry(ctx, cp, id) + _, _ = p.fetchYencHeaders(ctx, seg, groups) return nil }) } + + // Store-only fetches: nothing in the parse reads them, so they run behind + // the wait the parse blocks on, detached from the caller's cancellation + // (which fires as soon as the awaited warm-up returns) but bounded. + var storeOnly []string + if primary := p.primaryVideoToWarm(files); primary >= 0 { + storeOnly = append(storeOnly, files[primary].Segments[0].ID) + } + storeOnly = append(storeOnly, p.sevenZipTailToWarm(files)...) + if len(storeOnly) > 0 { + bg, cancel := context.WithTimeout(context.WithoutCancel(ctx), storeOnlyWarmTimeout) + go func() { + defer cancel() + fill := concpool.New().WithMaxGoroutines(maxFetch).WithContext(bg) + for _, id := range storeOnly { + fill.Go(func(ctx context.Context) error { + _, _ = p.fetchBodyWithRetry(ctx, cp, id) + return nil + }) + } + _ = fill.Wait() + }() + } _ = warm.Wait() } +// storeOnlyWarmTimeout bounds the detached warm fetches that only fill the +// segment store. +const storeOnlyWarmTimeout = 20 * time.Second + +// representativeMiddleSegment mirrors pickRepresentativeMiddleSegment on the +// raw NZB: the second segment of the first file with at least three, skipping +// files whose first segment is a declared gap. The parse re-derives its own +// choice from what it actually fetched; this only warms the likely answer. +func representativeMiddleSegment(files []nzbparser.NzbFile) (nzbparser.NzbSegment, []string, bool) { + for i := range files { + f := &files[i] + if len(f.Segments) < 3 || holes.IsPlaceholderID(f.Segments[0].ID) || holes.IsPlaceholderID(f.Segments[1].ID) { + continue + } + return f.Segments[1], f.Groups, true + } + return nzbparser.NzbSegment{}, nil, false +} + // sevenZipTailToWarm is the last two segment ids of the highest-numbered // .7z.NNN volume, or nil. 7z analysis opens the archive by reading the first // volume's head and then the end header at the very end of the last volume; diff --git a/internal/importer/parser/parser_warm_store_test.go b/internal/importer/parser/parser_warm_store_test.go index 5f0f8f8a1..5dd428af8 100644 --- a/internal/importer/parser/parser_warm_store_test.go +++ b/internal/importer/parser/parser_warm_store_test.go @@ -5,10 +5,13 @@ import ( "context" "sync" "testing" + "time" "github.com/javi11/nntppool/v4" "github.com/javi11/nzbparser" + "github.com/kipsilabs/altmount/internal/importer/parser/par2" "github.com/kipsilabs/altmount/internal/testsupport/fakepool" + "github.com/kipsilabs/altmount/internal/testsupport/par2gen" ) type recordingStore struct { @@ -16,7 +19,12 @@ type recordingStore struct { puts map[string][]byte } -func (s *recordingStore) Get(string) ([]byte, bool) { return nil, false } +func (s *recordingStore) Get(id string) ([]byte, bool) { + s.mu.Lock() + defer s.mu.Unlock() + d, ok := s.puts[id] + return d, ok +} func (s *recordingStore) Put(id string, data []byte) error { s.mu.Lock() defer s.mu.Unlock() @@ -83,6 +91,22 @@ func TestWarmFirstSegmentsWithoutStoreStillWarmsHeads(t *testing.T) { } } + +// waitForStore waits for a detached store-only warm fetch to land. +func waitForStore(t *testing.T, store *recordingStore, id string) []byte { + t.Helper() + deadline := time.Now().Add(4 * time.Second) + for { + if d, ok := store.get(id); ok { + return d + } + if time.Now().After(deadline) { + t.Fatalf("%s never landed in the segment store", id) + } + time.Sleep(10 * time.Millisecond) + } +} + func cleanVideos() nzbparser.NzbFiles { seg := func(prefix string) nzbparser.NzbSegments { return nzbparser.NzbSegments{ @@ -109,12 +133,10 @@ func TestWarmFirstSegmentsWarmsLargestVideoWhenStoreIsWired(t *testing.T) { p.WarmFirstSegments(context.Background(), cleanVideos()) + waitForStore(t, store, "e2-0") if got := fp.PerMessageCalls("e2-0"); got != 1 { t.Fatalf("largest video first segment fetched %d times, want 1", got) } - if _, ok := store.get("e2-0"); !ok { - t.Fatal("largest video first segment not put into the segment store") - } if got := fp.PerMessageCalls("e1-0"); got != 0 { t.Fatalf("smaller video first segment fetched %d times, want 0: only the largest is warmed", got) } @@ -161,12 +183,10 @@ func TestWarmFirstSegmentsWarmsLastSevenZipVolumeTail(t *testing.T) { p.WarmFirstSegments(context.Background(), sevenZipVolumes()) for _, id := range []string{"v3-3", "v3-2"} { + waitForStore(t, store, id) if got := fp.PerMessageCalls(id); got != 1 { t.Errorf("tail article %s fetched %d times, want 1", id, got) } - if _, ok := store.get(id); !ok { - t.Errorf("tail article %s not put into the segment store", id) - } } if got := fp.PerMessageCalls("v2-3") + fp.PerMessageCalls("v1-3") + fp.PerMessageCalls("v3-1"); got != 0 { t.Fatalf("unrelated articles fetched %d times, want 0", got) @@ -184,3 +204,80 @@ func TestWarmFirstSegmentsSkipsSevenZipTailWithoutStore(t *testing.T) { t.Fatalf("tail articles fetched %d times without a store, want 0", got) } } + +// The parse pass fetches one "representative" middle segment's yEnc header to +// learn the release-wide part size; started only after every first segment is +// in, it is a whole provider round trip on the critical path. Warm-up runs +// alongside the fast-fail probe, so fetching it there makes the parse a cache hit. +func TestWarmFirstSegmentsPrefetchesRepresentativeMiddleHeader(t *testing.T) { + fp := fakepool.New() + fp.SetDefaultBehavior(fakepool.SegmentBehavior{Bytes: []byte("v"), YEnc: nntppool.YEncMeta{FileSize: 3, PartSize: 1, Part: 2}}) + files := nzbparser.NzbFiles{ + {Filename: "Some.Release-GRP.r00", Segments: nzbparser.NzbSegments{ + {Bytes: 720000, Number: 1, ID: "r00-0"}, {Bytes: 720000, Number: 2, ID: "r00-1"}, {Bytes: 720000, Number: 3, ID: "r00-2"}, + }}, + } + p := NewParser(newFakeFullPoolManager(fp), stormConfigGetter(4)) + + p.WarmFirstSegments(context.Background(), files) + + if got := fp.PerMessageCalls("r00-1"); got != 1 { + t.Fatalf("representative middle segment fetched %d times during warm-up, want 1", got) + } + if head, ok := p.heads.get("r00-1"); !ok || head.meta.PartSize != 1 { + t.Fatalf("representative header not cached (ok=%v, %+v)", ok, head.meta) + } +} + +// Fetches that exist only to fill the segment store (the largest video's first +// article, the 7z tail) must not hold up the parse: WarmFirstSegments returns +// once the heads the parse needs are in, and those fetches finish behind it. +func TestWarmFirstSegmentsDoesNotWaitForStoreOnlyFetches(t *testing.T) { + fp := fakepool.New() + fp.SetDefaultBehavior(fakepool.SegmentBehavior{Bytes: []byte("v"), YEnc: nntppool.YEncMeta{FileSize: 1, PartSize: 1}}) + fp.SetBehavior("e2-0", fakepool.SegmentBehavior{Bytes: []byte("slow"), YEnc: nntppool.YEncMeta{FileSize: 1, PartSize: 1}, Latency: 1500 * time.Millisecond}) + store := &recordingStore{} + p := NewParser(newFakeFullPoolManager(fp), stormConfigGetter(4)) + p.SetSegmentStore(func() SegmentStore { return store }) + + start := time.Now() + p.WarmFirstSegments(context.Background(), cleanVideos()) + if elapsed := time.Since(start); elapsed > 700*time.Millisecond { + t.Fatalf("WarmFirstSegments blocked %s on a store-only fetch", elapsed) + } + deadline := time.Now().Add(4 * time.Second) + for { + if _, ok := store.get("e2-0"); ok { + break + } + if time.Now().After(deadline) { + t.Fatal("store-only fetch never landed in the segment store") + } + time.Sleep(20 * time.Millisecond) + } +} + +// The PAR2 index's first segment is fetched by the warm-up and published to the +// segment store; descriptor extraction reading it back from there costs no +// provider round trip. +func TestGetFileDescriptorsReadsIndexFromSegmentStore(t *testing.T) { + set := par2gen.BuildFull(1024, []par2gen.FileEntry{ + {Name: "Movie.mkv", Content: bytes.Repeat([]byte("m"), 4096)}, + }, 1) + fp := fakepool.New() // no behaviour for idx-0: a fetch would fail + store := &recordingStore{} + _ = store.Put("idx-0", set.Index) + file := &nzbparser.NzbFile{Filename: "Movie.par2", Segments: nzbparser.NzbSegments{{Bytes: len(set.Index), Number: 1, ID: "idx-0"}}} + cache := []*par2.FirstSegmentData{{File: file, RawBytes: set.Index[:min(len(set.Index), 16*1024)]}} + + descriptors, err := par2.GetFileDescriptors(context.Background(), cache, newFakeFullPoolManager(fp), store) + if err != nil { + t.Fatalf("GetFileDescriptors error = %v", err) + } + if len(descriptors) != 1 { + t.Fatalf("descriptors = %d, want 1", len(descriptors)) + } + if got := fp.PerMessageCalls("idx-0"); got != 0 { + t.Fatalf("index segment fetched %d times, want 0: it was in the segment store", got) + } +} From 518c4e65c3c1ed62530fb1d5faa919bbba9fb473 Mon Sep 17 00:00:00 2001 From: javi11 Date: Tue, 8 Sep 2026 20:47:51 +0200 Subject: [PATCH 09/11] perf(import): let the release probe pass with up to two unverified STATs of a full sample MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A healthy import sat for 4.5 s because one article of the 64 sampled stayed unanswered through two 2 s attempts — after the priority-lane hedge had been tried too, so the article is slow at the provider itself, not queued. The probe answers "is this post damaged?" from a sample; the thousands of articles it never sampled are handled at stream time, and so is this one. With every other sampled article healthy, the probe now passes after the first attempt when at most two of a sample of 32 or more remain unverified, logging what it left unchecked. Small samples and the per-file sweep — which maps exactly which files are broken — keep waiting. --- internal/importer/validation/fast_fail.go | 21 ++++++++ .../validation/fast_fail_hedge_test.go | 50 +++++++++++++++++-- 2 files changed, 68 insertions(+), 3 deletions(-) diff --git a/internal/importer/validation/fast_fail.go b/internal/importer/validation/fast_fail.go index ce7ece47a..2522a5605 100644 --- a/internal/importer/validation/fast_fail.go +++ b/internal/importer/validation/fast_fail.go @@ -136,6 +136,17 @@ func statIDsWithBoundedRetries( if len(remaining) == 0 { return missing, nil, nil } + if stopOnMissing && len(missing) == 0 && len(remaining) <= tolerableUnverified(len(ids)) { + // The release probe answers "is this post damaged?" from a + // sample. With everything else healthy, an article whose STAT + // neither the original request nor the priority hedge could get + // answered inside the ceiling is slow at the provider itself; + // waiting out further attempts held healthy imports for seconds. + // It is handled at stream time like the articles never sampled. + slog.InfoContext(ctx, "Fast-fail release probe proceeding with unverified stragglers", + "unverified", len(remaining), "sampled", len(ids), "attempt", attempt) + return missing, remaining, nil + } 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 @@ -179,6 +190,16 @@ const ( deadReleaseMissFraction = 0.5 ) +// tolerableUnverified is how many sampled articles the release probe may +// leave unanswered and still pass: two of a full 64-article sample, none of a +// small one, where each article is a large share of the evidence. +func tolerableUnverified(sampled int) int { + if sampled >= 32 { + return 2 + } + return 0 +} + // releaseLooksDead reports whether the definitive STAT answers collected so // far (missing out of reported) already prove the release unservable. A // release this damaged fails the holes policy regardless of how the diff --git a/internal/importer/validation/fast_fail_hedge_test.go b/internal/importer/validation/fast_fail_hedge_test.go index 071bfa45f..8a500693d 100644 --- a/internal/importer/validation/fast_fail_hedge_test.go +++ b/internal/importer/validation/fast_fail_hedge_test.go @@ -2,6 +2,7 @@ package validation import ( "context" + "errors" "fmt" "sync" "testing" @@ -16,7 +17,10 @@ import ( type delayedStatClient struct { *scriptedStatClient firstDelay map[string]time.Duration - sweeps int + // alwaysDelay applies to every STAT of the id, hedges included: an + // article that is slow at the provider, not one queued on a connection. + alwaysDelay map[string]time.Duration + sweeps int } func newDelayedStatClient(outcomes map[string][]error, firstDelay map[string]time.Duration) *delayedStatClient { @@ -44,8 +48,8 @@ func (c *delayedStatClient) StatMany(ctx context.Context, ids []string, _ nntppo if len(sequence) > 0 { err = sequence[min(attempt, len(sequence)-1)] } - delay := time.Duration(0) - if attempt == 0 { + delay := c.alwaysDelay[id] + if attempt == 0 && delay == 0 { delay = c.firstDelay[id] } c.mu.Unlock() @@ -231,3 +235,43 @@ func TestFastFailReleaseProbeHedgesWhenArrivalsStall(t *testing.T) { t.Fatalf("StatMany sweeps = %d, want 2 (primary + one hedge for every outstanding id)", got) } } + +// One article of sixty-four that neither the original STAT nor the priority +// hedge can get an answer for is slow at the provider itself; three 2 s +// attempts on it held a healthy import for 4.5 s in the bench. The release +// probe answers "damaged?" from a sample, and the articles it never sampled +// are handled at stream time — so is this one. The per-file sweep, which maps +// exactly which files are broken, keeps waiting. +func TestFastFailReleaseProbeToleratesAFewUnverifiedStragglers(t *testing.T) { + client := newDelayedStatClient(nil, nil) + client.alwaysDelay = map[string]time.Duration{"seg-40": 10 * 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 || missing { + t.Fatalf("FastFailReleaseProbe = (%v, %v), want (false, nil): 63 of 64 answered healthy", missing, err) + } + if elapsed > 2500*time.Millisecond { + t.Fatalf("probe took %s, want one attempt (the 2 s ceiling), not retries on the one slow article", elapsed) + } + if got := client.callCount("seg-40"); got > 2 { + t.Fatalf("slow article STATs = %d, want at most 2 (original + hedge)", got) + } +} + +func TestFastFailReleaseProbeDoesNotTolerateManyUnverified(t *testing.T) { + client := newDelayedStatClient(nil, nil) + client.alwaysDelay = map[string]time.Duration{} + for i := 40; i < 44; i++ { + client.alwaysDelay[fmt.Sprintf("seg-%d", i)] = 10 * time.Second + } + prev := fastFailStatBudget + fastFailStatBudget = 3 * time.Second + t.Cleanup(func() { fastFailStatBudget = prev }) + + _, err := FastFailReleaseProbe(context.Background(), probeFile(64), fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil) + if !errors.Is(err, ErrFastFailInconclusive) { + t.Fatalf("FastFailReleaseProbe error = %v, want ErrFastFailInconclusive: four unanswered is not a tolerable tail", err) + } +} From d61ac495fad3a44acca32aadc963d2a6e21bb24e Mon Sep 17 00:00:00 2001 From: javi11 Date: Tue, 8 Sep 2026 21:12:57 +0200 Subject: [PATCH 10/11] perf(import): judge a dead post from an 8-article first wave and skip its per-file sweep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three dead releases in a row cost the imports after them: each was STAT-ed 64 times over three attempts and then swept per file, hundreds of STATs the provider answers as slow 430 spool lookups that stay pipelined on the connections — the next healthy import's probe queued behind them and came back inconclusive or took 14 s. Eight sampled articles all missing is already the verdict. The release probe now checks a first wave of 8 (the edges plus random middle articles) to completion; when those misses condemn the release (releaseLooksDead) every file is marked broken without the sweep, otherwise the rest of the sample is checked as before. A healthy post pays one extra STAT round trip. --- internal/importer/processor.go | 19 ++- internal/importer/validation/fast_fail.go | 65 +------- .../validation/fast_fail_deadwave_test.go | 91 ++++++++++ .../validation/fast_fail_hedge_test.go | 18 +- .../importer/validation/fast_fail_verdict.go | 157 ++++++++++++++++++ 5 files changed, 275 insertions(+), 75 deletions(-) create mode 100644 internal/importer/validation/fast_fail_deadwave_test.go create mode 100644 internal/importer/validation/fast_fail_verdict.go diff --git a/internal/importer/processor.go b/internal/importer/processor.go index 29cc7399a..ccc41e192 100644 --- a/internal/importer/processor.go +++ b/internal/importer/processor.go @@ -432,7 +432,7 @@ func (proc *Processor) preParseFastFail(ctx context.Context, n *nzbparser.Nzb, c // common case — pay only this and skip the per-file sweep entirely, keeping // the "Checking segment availability" stage short. probeStart := time.Now() - missing, err := validation.FastFailReleaseProbe( + verdict, err := validation.FastFailReleaseProbeVerdict( ctx, fastFailFiles, proc.poolManager, @@ -444,10 +444,23 @@ func (proc *Processor) preParseFastFail(ctx context.Context, n *nzbparser.Nzb, c if err != nil { return nil, nil, nil, err } + missing := verdict.Missing acceptableMissingPercent := cfg.GetAcceptableMissingSegmentsPercentage() var results []validation.FastFailFileResult - if !missing { + switch { + case verdict.Dead: + // Every sampled article of the first wave is gone: the per-file sweep + // would only re-learn that with hundreds more STATs, each a slow 430 + // lookup left pipelined on the connections for the next import to + // queue behind. + if proc.log != nil { + proc.log.InfoContext(ctx, "Fast-fail release probe judged the release dead; skipping the per-file sweep", + "files", len(fastFailFiles), + "probe_duration", time.Since(probeStart)) + } + results = validation.DeadReleaseResults(fastFailFiles, verdict.MissingIDs) + case !missing: // The provider has everything the NZB lists. Gaps the NZB itself // declares are mapped from their placeholders without a STAT; the // per-file sweep would only re-learn what the probe just answered. @@ -465,7 +478,7 @@ func (proc *Processor) preParseFastFail(ctx context.Context, n *nzbparser.Nzb, c "files", len(fastFailFiles), "duration", time.Since(probeStart)) } - } else { + default: isStremioImport := (category != nil && *category == "stremio") || (downloadID != nil && strings.HasPrefix(*downloadID, "stremio:")) if isStremioImport && cfg.Stremio.EffectiveFastFailHeaderOnly() { if proc.log != nil { diff --git a/internal/importer/validation/fast_fail.go b/internal/importer/validation/fast_fail.go index 2522a5605..c23fd69b3 100644 --- a/internal/importer/validation/fast_fail.go +++ b/internal/importer/validation/fast_fail.go @@ -12,7 +12,6 @@ import ( metapb "github.com/kipsilabs/altmount/internal/metadata/proto" "github.com/kipsilabs/altmount/internal/pool" "github.com/kipsilabs/altmount/internal/progress" - "github.com/kipsilabs/altmount/internal/usenet" "github.com/javi11/nntppool/v4" ) @@ -359,68 +358,8 @@ func FastFailReleaseProbe( timeout time.Duration, patchIdx PatchIndex, ) (bool, error) { - var segments []*metapb.SegmentData - for _, file := range files { - for _, segment := range file.Segments { - if segment == nil || segment.Id == "" { - continue - } - if holes.IsPlaceholderID(segment.Id) { - // A gap the NZB itself declares is not a provider miss: it is - // mapped without a STAT (PlaceholderResults), and the probe's - // job is still to answer for the articles the NZB does list. - continue - } - segments = append(segments, segment) - } - } - if len(segments) == 0 { - return false, nil - } - - selected := capReleaseProbeSample(usenet.SelectSegmentsForValidation(segments, segmentSamplePercentage)) - if len(selected) == 0 { - return false, nil - } - - if !poolManager.HasPool() { - return false, fmt.Errorf("cannot fast-fail import: usenet connection pool is nil") - } - - usenetPool, err := poolManager.GetPool() - if err != nil { - return false, fmt.Errorf("cannot fast-fail import: usenet connection pool unavailable: %w", err) - } - if usenetPool == nil { - return false, fmt.Errorf("cannot fast-fail import: usenet connection pool is nil") - } - - if maxConnections <= 0 { - maxConnections = 1 - } - - ids := make([]string, len(selected)) - for i, seg := range selected { - ids[i] = seg.Id - } - - // Stat the sample via a bulk sweep, cancelling the rest on the first - // definitive miss. Operational errors retry only the affected IDs. Cap each - // attempt's probe timeout to 2 seconds per item so dead releases stay bounded. - probeTimeout := timeout - if probeTimeout > 2*time.Second { - probeTimeout = 2 * time.Second - } - missing, _, err := statIDsWithBoundedRetries(ctx, usenetPool, ids, maxConnections, probeTimeout, true, patchIdx) - if err != nil { - if len(missing) > 0 { - // The probe found a definitive miss before running out of - // patience for the rest; the answer is "damaged" either way. - return true, nil - } - return false, err - } - return len(missing) > 0, nil + v, err := FastFailReleaseProbeVerdict(ctx, files, poolManager, segmentSamplePercentage, maxConnections, timeout, patchIdx) + return v.Missing, err } // FastFailFileResult records the reachability outcome for a single FastFailFile. diff --git a/internal/importer/validation/fast_fail_deadwave_test.go b/internal/importer/validation/fast_fail_deadwave_test.go new file mode 100644 index 000000000..7ef6bf6cb --- /dev/null +++ b/internal/importer/validation/fast_fail_deadwave_test.go @@ -0,0 +1,91 @@ +package validation + +import ( + "context" + "testing" + "time" + + "github.com/javi11/nntppool/v4" +) + +// A dead post used to be STAT-ed 64 times, three attempts over, then swept per +// file: hundreds of slow 430 lookups left pipelined on the connections, which +// the next import's STATs then queued behind. Eight sampled articles all +// missing is already the verdict. +func TestFastFailReleaseProbeVerdictJudgesDeadPostFromFirstWave(t *testing.T) { + outcomes := make(map[string][]error, 64) + for _, seg := range makeTestSegments("seg", 64) { + outcomes[seg.Id] = []error{nntppool.ErrArticleNotFound} + } + client := newScriptedStatClient(outcomes) + + v, err := FastFailReleaseProbeVerdict(context.Background(), probeFile(64), fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil) + if err != nil { + t.Fatalf("FastFailReleaseProbeVerdict error = %v", err) + } + if !v.Missing || !v.Dead { + t.Fatalf("verdict = %+v, want Missing and Dead", v) + } + total := 0 + for _, seg := range makeTestSegments("seg", 64) { + total += client.callCount(seg.Id) + } + if total > probeFirstWave { + t.Fatalf("STATs issued = %d, want at most the first wave of %d", total, probeFirstWave) + } + if len(v.MissingIDs) == 0 { + t.Fatal("verdict carries no missing ids for the caller to record") + } +} + +func TestFastFailReleaseProbeVerdictHealthyPostChecksWholeSample(t *testing.T) { + client := newScriptedStatClient(nil) + + v, err := FastFailReleaseProbeVerdict(context.Background(), probeFile(64), fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil) + if err != nil || v.Missing || v.Dead { + t.Fatalf("verdict = %+v, err = %v, want healthy", v, err) + } + total := 0 + for _, seg := range makeTestSegments("seg", 64) { + total += client.callCount(seg.Id) + } + if total != 64 { + t.Fatalf("STATs issued = %d, want the whole 64-article sample on a healthy post", total) + } +} + +func TestFastFailReleaseProbeVerdictPartialDamageIsNotDead(t *testing.T) { + client := newScriptedStatClient(map[string][]error{"seg-1": {nntppool.ErrArticleNotFound}}) + + v, err := FastFailReleaseProbeVerdict(context.Background(), probeFile(64), fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil) + if err != nil { + t.Fatalf("FastFailReleaseProbeVerdict error = %v", err) + } + if !v.Missing || v.Dead { + t.Fatalf("verdict = %+v, want Missing but not Dead: one miss of eight is damage to map, not a dead post", v) + } +} + +func TestDeadReleaseResultsMarkEveryFileBroken(t *testing.T) { + files := []FastFailFile{ + {Filename: "a.part01.rar", Segments: makeTestSegments("a", 3), GroupKey: "a"}, + {Filename: "a.part02.rar", Segments: makeTestSegments("b", 3), GroupKey: "a"}, + {Filename: "a.par2"}, + } + results := DeadReleaseResults(files, []string{"a-0", "b-2"}) + if len(results) != 3 { + t.Fatalf("results = %d, want one per file", len(results)) + } + if !results[0].Broken || !results[1].Broken { + t.Fatalf("files with segments not marked broken: %+v", results[:2]) + } + if results[2].Broken { + t.Fatal("segment-less sidecar marked broken") + } + if got := results[0].MissingSegmentIDs; len(got) != 1 || got[0] != "a-0" { + t.Fatalf("file 0 missing ids = %v, want [a-0]", got) + } + if got := results[1].MissingSegmentIDs; len(got) != 1 || got[0] != "b-2" { + t.Fatalf("file 1 missing ids = %v, want [b-2]", got) + } +} diff --git a/internal/importer/validation/fast_fail_hedge_test.go b/internal/importer/validation/fast_fail_hedge_test.go index 8a500693d..0655c3544 100644 --- a/internal/importer/validation/fast_fail_hedge_test.go +++ b/internal/importer/validation/fast_fail_hedge_test.go @@ -129,8 +129,8 @@ func TestFastFailReleaseProbeDoesNotHedgeUniformlySlowSweep(t *testing.T) { 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) + if got := client.sweepCount(); got != 2 { + t.Fatalf("StatMany sweeps = %d, want 2 (first wave + rest): a uniformly slow provider has no stragglers to hedge", got) } } @@ -201,13 +201,13 @@ func TestFastFailReleaseProbeHedgesLargerStragglerTailOnPriorityLane(t *testing. } client.mu.Lock() defer client.mu.Unlock() - if len(client.opts) != 2 { - t.Fatalf("StatMany sweeps = %d, want 2 (primary + hedge)", len(client.opts)) + if len(client.opts) != 3 { + t.Fatalf("StatMany sweeps = %d, want 3 (first wave, rest, hedge)", len(client.opts)) } - if client.opts[0].Priority { - t.Fatal("primary sweep must stay on the normal lane") + if client.opts[0].Priority || client.opts[1].Priority { + t.Fatal("probe sweeps must stay on the normal lane") } - if !client.opts[1].Priority { + if !client.opts[2].Priority { t.Fatal("hedge sweep must use the priority lane") } } @@ -231,8 +231,8 @@ func TestFastFailReleaseProbeHedgesWhenArrivalsStall(t *testing.T) { if elapsed := time.Since(start); elapsed > 1500*time.Millisecond { t.Fatalf("probe took %s, want the 22 stalled STATs hedged inside the 2 s ceiling", elapsed) } - if got := client.sweepCount(); got != 2 { - t.Fatalf("StatMany sweeps = %d, want 2 (primary + one hedge for every outstanding id)", got) + if got := client.sweepCount(); got != 3 { + t.Fatalf("StatMany sweeps = %d, want 3 (first wave, rest, one hedge for every outstanding id)", got) } } diff --git a/internal/importer/validation/fast_fail_verdict.go b/internal/importer/validation/fast_fail_verdict.go new file mode 100644 index 000000000..c0ad225a9 --- /dev/null +++ b/internal/importer/validation/fast_fail_verdict.go @@ -0,0 +1,157 @@ +package validation + +import ( + "context" + "fmt" + "log/slog" + "time" + + "github.com/kipsilabs/altmount/internal/holes" + metapb "github.com/kipsilabs/altmount/internal/metadata/proto" + "github.com/kipsilabs/altmount/internal/pool" + "github.com/kipsilabs/altmount/internal/usenet" +) + +// probeFirstWave is how many sampled articles the release probe checks before +// committing the rest of the sample. A dead post — every sampled article gone — +// is the verdict after this many, and stopping there matters more than the +// one extra round trip a healthy post pays: a missing article costs the +// provider a slow spool lookup, and dozens of those left pipelined on the +// connections are what the next import's STATs queue behind. +const probeFirstWave = 8 + +// ProbeVerdict is what the release probe learned from its sample. +type ProbeVerdict struct { + // Missing reports a definitive 430/423 on at least one sampled article. + Missing bool + // Dead reports that the first wave alone condemned the release (see + // releaseLooksDead): the caller need not map which files are broken. + Dead bool + // MissingIDs are the sampled articles found missing. + MissingIDs []string +} + +// FastFailReleaseProbeVerdict runs the release probe in two waves: a small +// first wave whose misses can already prove the post dead, then — only when +// that wave was clean — the rest of the sample, cancelled on the first miss. +// Errors and the inconclusive rules are those of statIDsWithBoundedRetries. +func FastFailReleaseProbeVerdict( + ctx context.Context, + files []FastFailFile, + poolManager pool.Manager, + segmentSamplePercentage int, + maxConnections int, + timeout time.Duration, + patchIdx PatchIndex, +) (ProbeVerdict, error) { + var segments []*metapb.SegmentData + for _, file := range files { + for _, segment := range file.Segments { + if segment == nil || segment.Id == "" || holes.IsPlaceholderID(segment.Id) { + continue + } + segments = append(segments, segment) + } + } + if len(segments) == 0 { + return ProbeVerdict{}, nil + } + + selected := capReleaseProbeSample(usenet.SelectSegmentsForValidation(segments, segmentSamplePercentage)) + if len(selected) == 0 { + return ProbeVerdict{}, nil + } + + if !poolManager.HasPool() { + return ProbeVerdict{}, fmt.Errorf("cannot fast-fail import: usenet connection pool is nil") + } + usenetPool, err := poolManager.GetPool() + if err != nil { + return ProbeVerdict{}, fmt.Errorf("cannot fast-fail import: usenet connection pool unavailable: %w", err) + } + if usenetPool == nil { + return ProbeVerdict{}, fmt.Errorf("cannot fast-fail import: usenet connection pool is nil") + } + if maxConnections <= 0 { + maxConnections = 1 + } + + // Cap each attempt's probe timeout to 2 seconds per item so dead releases + // stay bounded. + probeTimeout := min(timeout, 2*time.Second) + + ids := make([]string, len(selected)) + for i, seg := range selected { + ids[i] = seg.Id + } + first, rest := ids, []string(nil) + if len(ids) > probeFirstWave { + first, rest = ids[:probeFirstWave], ids[probeFirstWave:] + } + + // The first wave is swept to completion, not cancelled on the first miss: + // its misses are counted to tell a dead post from a damaged one. + missing, unverified, err := statIDsWithBoundedRetries(ctx, usenetPool, first, maxConnections, probeTimeout, false, patchIdx) + if err != nil && len(missing) == 0 { + return ProbeVerdict{}, err + } + if len(missing) > 0 { + v := ProbeVerdict{Missing: true, MissingIDs: keys(missing)} + reported := len(first) - len(unverified) + if releaseLooksDead(len(missing), reported) { + v.Dead = true + slog.InfoContext(ctx, "Fast-fail release probe judged the release dead from its first wave", + "missing", len(missing), "sampled", len(first)) + } + return v, nil + } + if len(rest) == 0 { + return ProbeVerdict{}, nil + } + + missing, _, err = statIDsWithBoundedRetries(ctx, usenetPool, rest, maxConnections, probeTimeout, true, patchIdx) + if err != nil { + if len(missing) > 0 { + // A definitive miss before running out of patience for the rest; + // the answer is "damaged" either way. + return ProbeVerdict{Missing: true, MissingIDs: keys(missing)}, nil + } + return ProbeVerdict{}, err + } + return ProbeVerdict{Missing: len(missing) > 0, MissingIDs: keys(missing)}, nil +} + +// DeadReleaseResults is the per-file outcome of a release the probe judged +// dead: every file with segments is broken, carrying whichever sampled misses +// were its own. Index-aligned with files, like FastFailCheckFiles' results. +func DeadReleaseResults(files []FastFailFile, missingIDs []string) []FastFailFileResult { + missing := make(map[string]struct{}, len(missingIDs)) + for _, id := range missingIDs { + missing[id] = struct{}{} + } + results := make([]FastFailFileResult, len(files)) + for i, file := range files { + if len(file.Segments) == 0 { + continue + } + r := FastFailFileResult{Broken: true, SampledCount: len(file.Segments)} + for _, seg := range file.Segments { + if seg == nil { + continue + } + if _, gone := missing[seg.Id]; gone { + r.MissingSegmentIDs = append(r.MissingSegmentIDs, seg.Id) + } + } + results[i] = r + } + return results +} + +func keys(m map[string]error) []string { + out := make([]string, 0, len(m)) + for id := range m { + out = append(out, id) + } + return out +} From 445cddbd14f6291c078bb8fa2cc04b5383ad258a Mon Sep 17 00:00:00 2001 From: javi11 Date: Tue, 8 Sep 2026 21:40:30 +0200 Subject: [PATCH 11/11] perf(import): end a sweep attempt as soon as its answers condemn the release; open with a small chunk MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A damaged post (first article of every volume gone) reaches the per-file sweep, which dispatched 64 STATs and waited the attempt out — 18 s on a provider answering 430s slowly — although the first dozen misses had already condemned it, and every 430 left in flight was a slow spool lookup pipelined on a connection for the next import's probe to queue behind. A sweep attempt now cancels itself the moment the definitive answers meet releaseLooksDead, returning what it has so the existing dead-release path condemns the rest, and the sweep's first chunk is capped at 16 STATs so a dead post leaves few of them in flight. --- internal/importer/validation/fast_fail.go | 26 ++++++++++- .../validation/fast_fail_deadwave_test.go | 45 +++++++++++++++++++ 2 files changed, 69 insertions(+), 2 deletions(-) diff --git a/internal/importer/validation/fast_fail.go b/internal/importer/validation/fast_fail.go index c23fd69b3..f2d66e2dd 100644 --- a/internal/importer/validation/fast_fail.go +++ b/internal/importer/validation/fast_fail.go @@ -73,6 +73,8 @@ func statIDsWithBoundedRetries( 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)) + definitive := 0 + deadEarly := false for result := range hedgedStatMany(statCtx, client, remaining, maxConnections) { if _, wanted := seen[result.MessageID]; !wanted { @@ -80,16 +82,19 @@ func statIDsWithBoundedRetries( } reported[result.MessageID] = true if result.Err == nil { + definitive++ continue } // Repaired bytes live only in the local patch store, so an article // the providers dropped is still available. Reported with no error // recorded, it leaves the retry set as reachable. if patched(patchIdx, result.MessageID) { + definitive++ continue } if isDefinitiveFastFailMiss(result.Err) { missing[result.MessageID] = result.Err + definitive++ if stopOnMissing { if ctxErr := ctx.Err(); ctxErr != nil { cancel() @@ -98,6 +103,14 @@ func statIDsWithBoundedRetries( cancel() return missing, nil, nil } + if releaseLooksDead(len(missing), definitive) { + // The answers so far already condemn the release; the + // STATs still in flight are slow 430 lookups that cannot + // change it and only hold the import and the connections. + deadEarly = true + cancel() + break + } continue } transient[result.MessageID] = result.Err @@ -135,6 +148,10 @@ func statIDsWithBoundedRetries( if len(remaining) == 0 { return missing, nil, nil } + if deadEarly { + return missing, remaining, fmt.Errorf("%w: %d segment(s) left unverified once %d misses condemned the release", + ErrFastFailInconclusive, len(remaining), len(missing)) + } if stopOnMissing && len(missing) == 0 && len(remaining) <= tolerableUnverified(len(ids)) { // The release probe answers "is this post damaged?" from a // sample. With everything else healthy, an article whose STAT @@ -182,6 +199,11 @@ func statIDsWithBoundedRetries( // maxSweepChunk is the most STATs the per-file sweep has outstanding at once. const maxSweepChunk = 64 +// firstSweepChunk bounds the sweep's opening wave: a dead post is condemned +// by its first few misses, and every STAT past those is a slow 430 lookup left +// pipelined on a connection for the next import to queue behind. +const firstSweepChunk = 16 + // Dead-post thresholds for releaseLooksDead: at least this many definitive // misses, making up at least this share of the definitive answers so far. const ( @@ -523,8 +545,8 @@ func FastFailCheckFiles( // times out behind the backlog. Smaller waves let the dead-release verdict // fire after one wave with little left outstanding. chunkSize := min(maxConnections, maxSweepChunk) - for start := 0; start < total; start += chunkSize { - end := min(start+chunkSize, total) + for start, size := 0, min(chunkSize, firstSweepChunk); start < total; start, size = start+size, chunkSize { + end := min(start+size, total) chunk := jobs[start:end] toCheck := make([]statJob, 0, len(chunk)) diff --git a/internal/importer/validation/fast_fail_deadwave_test.go b/internal/importer/validation/fast_fail_deadwave_test.go index 7ef6bf6cb..0ff7a8c5e 100644 --- a/internal/importer/validation/fast_fail_deadwave_test.go +++ b/internal/importer/validation/fast_fail_deadwave_test.go @@ -2,6 +2,7 @@ package validation import ( "context" + "fmt" "testing" "time" @@ -89,3 +90,47 @@ func TestDeadReleaseResultsMarkEveryFileBroken(t *testing.T) { t.Fatalf("file 1 missing ids = %v, want [b-2]", got) } } + +// A damaged-not-dead post (first article of every volume gone) reaches the +// per-file sweep. On a provider answering 430s slowly the sweep used to +// dispatch 64 STATs and wait the attempt out — 18 s in the bench — although +// the first dozen misses already condemned the release, and every 430 it left +// in flight slowed the import that followed. The attempt ends as soon as the +// definitive answers prove the post dead, and the first chunk stays small. +func TestFastFailCheckFilesEndsAttemptOnceReleaseIsDead(t *testing.T) { + var files []FastFailFile + outcomes := make(map[string][]error, 64) + slow := make(map[string]time.Duration, 64) + for i := range 64 { + segs := makeTestSegments(fmt.Sprintf("v%02d", i), 1) + files = append(files, FastFailFile{Filename: fmt.Sprintf("vol%02d.mkv", i), Segments: segs}) + outcomes[segs[0].Id] = []error{nntppool.ErrArticleNotFound} + if i >= 12 { + slow[segs[0].Id] = 3 * time.Second + } + } + client := newDelayedStatClient(outcomes, nil) + client.alwaysDelay = slow + + start := time.Now() + results, err := FastFailCheckFiles(context.Background(), files, fastFailPoolManager{client: client}, 100, 64, 30*time.Second, nil, nil, false) + elapsed := time.Since(start) + if err != nil { + t.Fatalf("FastFailCheckFiles error = %v, want the dead-release verdict", err) + } + for i, r := range results { + if !r.Broken { + t.Fatalf("results[%d] not Broken on a dead release", i) + } + } + if elapsed > 1500*time.Millisecond { + t.Fatalf("sweep took %s, want the attempt cut once a dozen misses condemned the release", elapsed) + } + total := 0 + for i := range 64 { + total += client.callCount(fmt.Sprintf("v%02d-0", i)) + } + if total > firstSweepChunk { + t.Fatalf("STATs issued = %d, want at most the first chunk of %d", total, firstSweepChunk) + } +}