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

cmd/altmount/cmd/serve.go

@@ -142,6 +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(cacheSource.Store)
 
 	// Background PAR2 repair: repairs missing articles and serves the patched
 	// payloads on the read path's hole branch.

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,14 +20,34 @@ 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
 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)
 

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)
 

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
 }

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

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)