diff --git a/cmd/altmount/cmd/setup.go b/cmd/altmount/cmd/setup.go index 979b0b5f7..50846f61a 100644 --- a/cmd/altmount/cmd/setup.go +++ b/cmd/altmount/cmd/setup.go @@ -329,7 +329,7 @@ func initializeSegmentCache(ctx context.Context, cfg *config.Config, source *seg mgr.Start(ctx) source.Swap(mgr) - slog.InfoContext(ctx, "Segment cache initialized", + slog.InfoContext(ctx, "Segment cache started (catalog loads in background)", "cache_path", mgrCfg.CachePath, "max_size_bytes", mgrCfg.MaxSizeBytes, "expiry_duration", mgrCfg.ExpiryDuration) diff --git a/internal/nzbfilesystem/segcache/cache.go b/internal/nzbfilesystem/segcache/cache.go index 8fa12f479..e53d89d8d 100644 --- a/internal/nzbfilesystem/segcache/cache.go +++ b/internal/nzbfilesystem/segcache/cache.go @@ -42,9 +42,12 @@ type SegmentCache struct { logger *slog.Logger totalSize int64 dirty atomic.Bool + loading atomic.Bool } -// NewSegmentCache creates a new segment cache, loading any existing catalog. +// NewSegmentCache creates a new segment cache. It does NOT load any existing +// catalog; call LoadCatalog to hydrate from disk. Manager.Start runs that load +// in a background goroutine so the per-segment stat loop does not block boot. func NewSegmentCache(cfg Config, logger *slog.Logger) (*SegmentCache, error) { if err := os.MkdirAll(cfg.CachePath, 0o755); err != nil { return nil, fmt.Errorf("segcache: create cache dir %s: %w", cfg.CachePath, err) @@ -55,8 +58,6 @@ func NewSegmentCache(cfg Config, logger *slog.Logger) (*SegmentCache, error) { config: cfg, logger: logger, } - c.loadCatalog() - return c, nil } @@ -96,6 +97,13 @@ func (c *SegmentCache) Get(messageID string) ([]byte, bool) { // Put stores segment bytes atomically (temp-write + rename). func (c *SegmentCache) Put(messageID string, data []byte) error { + // Skip caching while the catalog hydrates. The segment is still served from + // Usenet, just not persisted this round; it gets cached on the next request + // after load. Avoids Put/load lock contention on boot. + if c.loading.Load() { + return nil + } + h := sha256.Sum256([]byte(messageID)) filename := hex.EncodeToString(h[:]) + ".seg" dataPath := filepath.Join(c.config.CachePath, filename) @@ -245,8 +253,17 @@ func (c *SegmentCache) SaveCatalog() error { return nil } -func (c *SegmentCache) loadCatalog() { +// LoadCatalog hydrates the in-memory catalog from catalog.json on disk, statting +// each .seg file and dropping entries whose data is missing. Put is gated off +// (see the loading flag) for the duration, so the load can assign the map +// wholesale without racing a concurrent writer. Sets the gate on entry and +// clears it on exit. +func (c *SegmentCache) LoadCatalog() { + c.loading.Store(true) + defer c.loading.Store(false) + catalogPath := filepath.Join(c.config.CachePath, "catalog.json") + c.logger.Info("segcache: loading catalog", "path", catalogPath) data, err := os.ReadFile(catalogPath) if err != nil { @@ -270,8 +287,10 @@ func (c *SegmentCache) loadCatalog() { } } + c.mu.Lock() c.items = valid c.totalSize = totalSize + c.mu.Unlock() c.logger.Info("segcache: catalog loaded", "items", len(valid), "total_bytes", totalSize) } diff --git a/internal/nzbfilesystem/segcache/cache_internal_test.go b/internal/nzbfilesystem/segcache/cache_internal_test.go new file mode 100644 index 000000000..9caba23e0 --- /dev/null +++ b/internal/nzbfilesystem/segcache/cache_internal_test.go @@ -0,0 +1,32 @@ +package segcache + +import ( + "log/slog" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestPutIgnoredWhileLoading guards the loading gate that replaced the +// merge-under-lock path: while the catalog hydrates, Put must be a silent no-op +// so a segment downloaded during cold load neither contends with the load nor +// gets clobbered by the wholesale map assignment in LoadCatalog. Once the gate +// clears, Put works normally again. +func TestPutIgnoredWhileLoading(t *testing.T) { + cfg := Config{CachePath: t.TempDir(), MaxSizeBytes: 10 * 1024 * 1024} + c, err := NewSegmentCache(cfg, slog.Default()) + require.NoError(t, err) + + c.loading.Store(true) + + require.NoError(t, c.Put("busy@msg", []byte("data"))) + assert.False(t, c.Has("busy@msg"), "Put must be ignored while catalog is loading") + assert.EqualValues(t, 0, c.ItemCount()) + assert.EqualValues(t, 0, c.TotalSize()) + + c.loading.Store(false) + + require.NoError(t, c.Put("busy@msg", []byte("data"))) + assert.True(t, c.Has("busy@msg")) +} diff --git a/internal/nzbfilesystem/segcache/cache_test.go b/internal/nzbfilesystem/segcache/cache_test.go index 27857f198..3cea87878 100644 --- a/internal/nzbfilesystem/segcache/cache_test.go +++ b/internal/nzbfilesystem/segcache/cache_test.go @@ -22,6 +22,8 @@ func newTestCache(t *testing.T, maxBytes int64, expiry time.Duration) *segcache. } c, err := segcache.NewSegmentCache(cfg, slog.Default()) require.NoError(t, err) + // Hydrate + clear the loading gate; mirrors Manager.Start before serving Puts. + c.LoadCatalog() return c } @@ -93,12 +95,14 @@ func TestCacheSaveCatalogAndReload(t *testing.T) { // Write entries, save catalog, then reload. c1, err := segcache.NewSegmentCache(cfg, slog.Default()) require.NoError(t, err) + c1.LoadCatalog() require.NoError(t, c1.Put("persist@msg", []byte("persistent data"))) require.NoError(t, c1.SaveCatalog()) // Load a new cache from the same directory. c2, err := segcache.NewSegmentCache(cfg, slog.Default()) require.NoError(t, err) + c2.LoadCatalog() assert.True(t, c2.Has("persist@msg"), "reloaded cache should contain persisted entry") got, ok := c2.Get("persist@msg") @@ -129,6 +133,7 @@ func TestCacheCatalogSurvivesMissingSegFiles(t *testing.T) { c1, err := segcache.NewSegmentCache(cfg, slog.Default()) require.NoError(t, err) + c1.LoadCatalog() require.NoError(t, c1.Put("good@msg", []byte("good"))) require.NoError(t, c1.Put("bad@msg", []byte("will be deleted"))) require.NoError(t, c1.SaveCatalog()) @@ -152,7 +157,9 @@ func TestCacheCatalogSurvivesMissingSegFiles(t *testing.T) { // Reload: only "good@msg" should survive. c2, err := segcache.NewSegmentCache(cfg, slog.Default()) require.NoError(t, err) + c2.LoadCatalog() assert.True(t, c2.Has("good@msg")) assert.False(t, c2.Has("bad@msg"), "entry with missing seg file should be dropped on reload") } + diff --git a/internal/nzbfilesystem/segcache/file_test.go b/internal/nzbfilesystem/segcache/file_test.go index 53bc4ee5e..2daf63938 100644 --- a/internal/nzbfilesystem/segcache/file_test.go +++ b/internal/nzbfilesystem/segcache/file_test.go @@ -19,6 +19,8 @@ func newDefaultTestCache(t *testing.T) *segcache.SegmentCache { } cache, err := segcache.NewSegmentCache(cfg, slog.Default()) require.NoError(t, err) + // Hydrate + clear the loading gate; mirrors Manager.Start before serving Puts. + cache.LoadCatalog() return cache } diff --git a/internal/nzbfilesystem/segcache/manager.go b/internal/nzbfilesystem/segcache/manager.go index 9c5ddb934..967a72b60 100644 --- a/internal/nzbfilesystem/segcache/manager.go +++ b/internal/nzbfilesystem/segcache/manager.go @@ -86,9 +86,15 @@ func NewManager(cfg ManagerConfig, logger *slog.Logger) (*Manager, error) { }, nil } -// Start launches background maintenance goroutines. +// Start launches background maintenance goroutines, including the initial +// catalog load (run here, not in NewManager, so boot is not blocked by the +// per-segment stat loop). Stop waits for all of them. func (m *Manager) Start(_ context.Context) { - m.wg.Add(2) + m.wg.Add(3) + go func() { + defer m.wg.Done() + m.cache.LoadCatalog() + }() go m.cleanupLoop() go m.catalogFlushLoop() }