fix: hold back JetStream recovery headroom on the store filesystem (#967)

Filling the filesystem behind the JetStream store to zero free bytes leaves nats-server with nowhere to write while it recovers a stream. Every error return inside the locked region of newFileStoreWithCreated assigns nil to the named return value fs, and the deferred unlock then dereferences it, so a recovery write failure surfaces as a nil-pointer panic instead of an error. That structure is unchanged in v2.14.3, in the pinned v2.14.5 and in the current v2.14.6, so no available version bump addresses it. The nats service now keeps a preallocated reserve file beside the jetstream directory and hands it back at startup when the filesystem can no longer supply that headroom on its own, re-arming once space returns. Recovery therefore always has somewhere to write, whichever process filled the disk.
2026-10-19 10:52:04 +00:00 · 2026-09-03 14:46:11 +10:00
parent 18aa67b0e9
commit eccc61a69e
3 changed files with 452 additions and 0 deletions
@@ -97,6 +97,12 @@ func launchService(config *Config) (err error) {

 	slog.Debug("NATS server options", "opts", opts)

+	reserve := &StoreReserve{Dir: opts.StoreDir, Size: DefaultStoreReserveBytes}
+	if !opts.JetStream {
+		reserve.Size = 0
+	}
+	reserve.Apply()
+
 	// Initialize new server with options
 	ns, err := server.NewServer(opts)
 	if err != nil {
@@ -110,6 +116,10 @@ func launchService(config *Config) (err error) {
 		server.PrintAndDie(err.Error())
 	}

+	stop := make(chan struct{})
+	defer close(stop)
+	go reserve.Maintain(stop)
+
 	ns.WaitForShutdown()

 	return nil
@@ -0,0 +1,167 @@
+package nats
+
+import (
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"syscall"
+	"time"
+)
+
+// DefaultStoreReserveBytes is the headroom held back on the JetStream store
+// filesystem. Stream recovery rewrites index, meta and tombstone state before
+// the server accepts writes, and 256 MiB covers that with margin.
+const DefaultStoreReserveBytes int64 = 256 << 20
+
+// DefaultStoreReserveInterval paces the background re-arm. The reserve only
+// changes state around a full disk, so sampling is cheap and rare.
+const DefaultStoreReserveInterval = 15 * time.Minute
+
+// StoreReserveName is the reserve file. It sits beside the "jetstream"
+// directory rather than inside it: nats-server enumerates the contents of
+// store_dir/jetstream as accounts, but never the reserve's own parent entries.
+const StoreReserveName = ".spinifex-nats-reserve"
+
+// StoreReserve is a preallocated file that keeps the JetStream store
+// filesystem off literal zero. nats-server dereferences a nil named return in
+// its deferred unlock when recovery cannot write, so a full disk kills it.
+type StoreReserve struct {
+	// Dir is the configured JetStream store_dir.
+	Dir string
+	// Size is how many bytes to hold back. Zero or negative disables the reserve.
+	Size int64
+	// Interval paces Maintain. Zero selects DefaultStoreReserveInterval.
+	Interval time.Duration
+	// FreeBytes reports bytes available to an unprivileged writer on Dir's
+	// filesystem. Nil selects statfs; tests substitute their own probe.
+	FreeBytes func(dir string) (uint64, error)
+}
+
+// Path returns the reserve file's location.
+func (r *StoreReserve) Path() string {
+	return filepath.Join(r.Dir, StoreReserveName)
+}
+
+// Prepare arms the reserve when the store filesystem has room to spare and
+// releases it when the filesystem can no longer supply recovery headroom on
+// its own. It reports whether this call released the reserve.
+func (r *StoreReserve) Prepare() (bool, error) {
+	if r.Size <= 0 || r.Dir == "" {
+		return false, nil
+	}
+	if err := os.MkdirAll(r.Dir, 0o700); err != nil {
+		return false, fmt.Errorf("create jetstream store dir %s: %w", r.Dir, err)
+	}
+	free, err := r.free()
+	if err != nil {
+		return false, err
+	}
+
+	// Arming costs Size, so only arm when what remains still clears Size.
+	// A single threshold would arm and release on alternate starts.
+	size := uint64(r.Size)
+	switch {
+	case free < size:
+		return r.release()
+	case free >= 2*size:
+		return false, r.arm()
+	default:
+		return false, nil
+	}
+}
+
+// Apply runs Prepare and reports the outcome, logging rather than failing:
+// a server with no headroom left must still be allowed to start.
+func (r *StoreReserve) Apply() {
+	released, err := r.Prepare()
+	if err != nil {
+		slog.Warn("JetStream store reserve unavailable", "store_dir", r.Dir, "err", err)
+		return
+	}
+	if released {
+		slog.Warn("Released the JetStream store reserve, the store filesystem is out of space",
+			"store_dir", r.Dir, "reserve_bytes", r.Size)
+	}
+}
+
+// Maintain re-arms the reserve once the filesystem recovers, so a node that
+// spent its headroom is protected again without waiting for the next restart.
+// It returns when stop is closed.
+func (r *StoreReserve) Maintain(stop <-chan struct{}) {
+	interval := r.Interval
+	if interval <= 0 {
+		interval = DefaultStoreReserveInterval
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-stop:
+			return
+		case <-ticker.C:
+			r.Apply()
+		}
+	}
+}
+
+// free reports the bytes available on the store filesystem, using the
+// injected probe when one is set.
+func (r *StoreReserve) free() (uint64, error) {
+	if r.FreeBytes != nil {
+		return r.FreeBytes(r.Dir)
+	}
+	return statfsFreeBytes(r.Dir)
+}
+
+// arm creates the reserve file at Size, allocating real blocks. It is a no-op
+// once the file is already the right size.
+func (r *StoreReserve) arm() error {
+	f, err := os.OpenFile(r.Path(), os.O_RDWR|os.O_CREATE, 0o600)
+	if err != nil {
+		return fmt.Errorf("open jetstream store reserve: %w", err)
+	}
+	defer f.Close()
+
+	fi, err := f.Stat()
+	if err != nil {
+		return fmt.Errorf("stat jetstream store reserve: %w", err)
+	}
+	if fi.Size() == r.Size {
+		return nil
+	}
+	// Fallocate, not Truncate: a sparse file reserves no blocks at all, so
+	// the guard would hold back nothing while appearing to be armed.
+	if err := syscall.Fallocate(int(f.Fd()), 0, 0, r.Size); err != nil {
+		return fmt.Errorf("allocate jetstream store reserve: %w", err)
+	}
+	return nil
+}
+
+// release removes the reserve file, handing its blocks back to the
+// filesystem so stream recovery has somewhere to write.
+func (r *StoreReserve) release() (bool, error) {
+	err := os.Remove(r.Path())
+	switch {
+	case err == nil:
+		return true, nil
+	case errors.Is(err, os.ErrNotExist):
+		return false, nil
+	default:
+		return false, fmt.Errorf("release jetstream store reserve: %w", err)
+	}
+}
+
+// statfsFreeBytes reports the space available to an unprivileged writer on the
+// filesystem holding dir, which is what a full disk actually leaves JetStream.
+func statfsFreeBytes(dir string) (uint64, error) {
+	var st syscall.Statfs_t
+	if err := syscall.Statfs(dir, &st); err != nil {
+		return 0, fmt.Errorf("statfs %s: %w", dir, err)
+	}
+	if st.Bsize <= 0 {
+		return 0, fmt.Errorf("statfs %s: invalid block size %d", dir, st.Bsize)
+	}
+	return st.Bavail * uint64(st.Bsize), nil
+}
@@ -0,0 +1,275 @@
+package nats_test
+
+import (
+	"errors"
+	"os"
+	"path/filepath"
+	"syscall"
+	"testing"
+	"time"
+
+	svcnats "github.com/mulgadc/spinifex/spinifex/services/nats"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const testReserveSize = 64 << 10
+
+// fixedFree returns a probe that reports a constant amount of free space,
+// which is the only way to drive the starved branch on a real filesystem.
+func fixedFree(free uint64) func(string) (uint64, error) {
+	return func(string) (uint64, error) { return free, nil }
+}
+
+func newTestReserve(t *testing.T, free uint64) *svcnats.StoreReserve {
+	t.Helper()
+	return &svcnats.StoreReserve{
+		Dir:       filepath.Join(t.TempDir(), "nats"),
+		Size:      testReserveSize,
+		FreeBytes: fixedFree(free),
+	}
+}
+
+// allocatedBytes reports the blocks actually charged to path, so a sparse
+// file that reserves nothing is distinguishable from a real allocation.
+func allocatedBytes(t *testing.T, path string) int64 {
+	t.Helper()
+	fi, err := os.Stat(path)
+	require.NoError(t, err)
+	st, ok := fi.Sys().(*syscall.Stat_t)
+	require.True(t, ok, "stat is not a syscall.Stat_t")
+	return st.Blocks * 512
+}
+
+func TestStoreReserve_PathSitsBesideJetStreamDir(t *testing.T) {
+	r := &svcnats.StoreReserve{Dir: "/var/lib/spinifex/nats"}
+	assert.Equal(t, "/var/lib/spinifex/nats/"+svcnats.StoreReserveName, r.Path())
+	assert.NotEqual(t, "jetstream", svcnats.StoreReserveName,
+		"the reserve must never be mistaken for the jetstream directory")
+}
+
+func TestStoreReserve_ArmsWhenSpaceIsPlentiful(t *testing.T) {
+	r := newTestReserve(t, 100*testReserveSize)
+
+	released, err := r.Prepare()
+	require.NoError(t, err)
+	assert.False(t, released)
+
+	fi, err := os.Stat(r.Path())
+	require.NoError(t, err)
+	assert.Equal(t, int64(testReserveSize), fi.Size())
+}
+
+func TestStoreReserve_ArmAllocatesRealBlocks(t *testing.T) {
+	r := newTestReserve(t, 100*testReserveSize)
+
+	_, err := r.Prepare()
+	require.NoError(t, err)
+
+	assert.GreaterOrEqual(t, allocatedBytes(t, r.Path()), int64(testReserveSize),
+		"a sparse reserve would hold back no space at all")
+}
+
+func TestStoreReserve_ArmIsIdempotent(t *testing.T) {
+	r := newTestReserve(t, 100*testReserveSize)
+
+	_, err := r.Prepare()
+	require.NoError(t, err)
+	before, err := os.Stat(r.Path())
+	require.NoError(t, err)
+
+	released, err := r.Prepare()
+	require.NoError(t, err)
+	assert.False(t, released)
+
+	after, err := os.Stat(r.Path())
+	require.NoError(t, err)
+	assert.Equal(t, before.Size(), after.Size())
+}
+
+func TestStoreReserve_ReleasesWhenStarved(t *testing.T) {
+	r := newTestReserve(t, 100*testReserveSize)
+	_, err := r.Prepare()
+	require.NoError(t, err)
+	require.FileExists(t, r.Path())
+
+	// The filesystem has hit zero: the reserve is the only headroom left for
+	// stream recovery, so it must be handed back.
+	r.FreeBytes = fixedFree(0)
+	released, err := r.Prepare()
+	require.NoError(t, err)
+	assert.True(t, released)
+	assert.NoFileExists(t, r.Path())
+}
+
+func TestStoreReserve_ReleaseWhenAlreadyAbsentIsNotReported(t *testing.T) {
+	r := newTestReserve(t, 0)
+
+	released, err := r.Prepare()
+	require.NoError(t, err)
+	assert.False(t, released, "an absent reserve was not released by this call")
+	assert.NoFileExists(t, r.Path())
+}
+
+func TestStoreReserve_HoldsSteadyOnMarginalFreeSpace(t *testing.T) {
+	// Between one and two reserves free: arming would drop below the
+	// threshold and release on the next start, so nothing should change.
+	r := newTestReserve(t, testReserveSize+1)
+
+	released, err := r.Prepare()
+	require.NoError(t, err)
+	assert.False(t, released)
+	assert.NoFileExists(t, r.Path())
+}
+
+func TestStoreReserve_RearmsAfterSpaceIsFreed(t *testing.T) {
+	r := newTestReserve(t, 0)
+	_, err := r.Prepare()
+	require.NoError(t, err)
+	require.NoFileExists(t, r.Path())
+
+	r.FreeBytes = fixedFree(100 * testReserveSize)
+	released, err := r.Prepare()
+	require.NoError(t, err)
+	assert.False(t, released)
+	assert.FileExists(t, r.Path())
+}
+
+func TestStoreReserve_DisabledWhenSizeIsZero(t *testing.T) {
+	r := newTestReserve(t, 0)
+	r.Size = 0
+
+	released, err := r.Prepare()
+	require.NoError(t, err)
+	assert.False(t, released)
+	assert.NoDirExists(t, r.Dir, "a disabled reserve must not touch the store dir")
+}
+
+func TestStoreReserve_DisabledWhenDirIsEmpty(t *testing.T) {
+	r := &svcnats.StoreReserve{Size: testReserveSize}
+
+	released, err := r.Prepare()
+	require.NoError(t, err)
+	assert.False(t, released)
+}
+
+func TestStoreReserve_CreatesMissingStoreDir(t *testing.T) {
+	r := newTestReserve(t, 100*testReserveSize)
+	require.NoDirExists(t, r.Dir)
+
+	_, err := r.Prepare()
+	require.NoError(t, err)
+	assert.DirExists(t, r.Dir)
+}
+
+func TestStoreReserve_ReportsProbeFailure(t *testing.T) {
+	sentinel := errors.New("probe failed")
+	r := newTestReserve(t, 0)
+	r.FreeBytes = func(string) (uint64, error) { return 0, sentinel }
+
+	released, err := r.Prepare()
+	assert.False(t, released)
+	require.ErrorIs(t, err, sentinel)
+}
+
+func TestStoreReserve_ReportsUncreatableStoreDir(t *testing.T) {
+	parent := t.TempDir()
+	blocker := filepath.Join(parent, "nats")
+	require.NoError(t, os.WriteFile(blocker, nil, 0o600))
+
+	r := &svcnats.StoreReserve{Dir: blocker, Size: testReserveSize, FreeBytes: fixedFree(0)}
+	released, err := r.Prepare()
+	assert.False(t, released)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "jetstream store dir")
+}
+
+// TestStoreReserve_DefaultProbeUsesStatfs exercises the production free-space
+// path, which the injected probe otherwise hides.
+func TestStoreReserve_DefaultProbeUsesStatfs(t *testing.T) {
+	r := &svcnats.StoreReserve{Dir: filepath.Join(t.TempDir(), "nats"), Size: testReserveSize}
+
+	released, err := r.Prepare()
+	require.NoError(t, err)
+	assert.False(t, released)
+	assert.FileExists(t, r.Path(), "a temp dir should have room for a 64 KiB reserve")
+}
+
+func TestStoreReserve_ApplyArmsAndSurvivesFailure(t *testing.T) {
+	r := newTestReserve(t, 100*testReserveSize)
+	r.Apply()
+	assert.FileExists(t, r.Path())
+
+	// A probe failure must be swallowed so the server can still start.
+	r.FreeBytes = func(string) (uint64, error) { return 0, errors.New("probe failed") }
+	assert.NotPanics(t, r.Apply)
+	assert.FileExists(t, r.Path())
+}
+
+func TestStoreReserve_ApplyReleasesWhenStarved(t *testing.T) {
+	r := newTestReserve(t, 100*testReserveSize)
+	r.Apply()
+	require.FileExists(t, r.Path())
+
+	r.FreeBytes = fixedFree(0)
+	r.Apply()
+	assert.NoFileExists(t, r.Path())
+}
+
+func TestStoreReserve_MaintainRearmsUntilStopped(t *testing.T) {
+	r := newTestReserve(t, 0)
+	r.Interval = time.Millisecond
+	_, err := r.Prepare()
+	require.NoError(t, err)
+	require.NoFileExists(t, r.Path())
+
+	stop := make(chan struct{})
+	done := make(chan struct{})
+	r.FreeBytes = fixedFree(100 * testReserveSize)
+	go func() {
+		r.Maintain(stop)
+		close(done)
+	}()
+
+	require.Eventually(t, func() bool {
+		_, err := os.Stat(r.Path())
+		return err == nil
+	}, 5*time.Second, 5*time.Millisecond, "Maintain should re-arm once space returns")
+
+	close(stop)
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Maintain did not return after stop was closed")
+	}
+}
+
+func TestStoreReserve_MaintainStopsBeforeFirstTick(t *testing.T) {
+	r := newTestReserve(t, 100*testReserveSize)
+	stop := make(chan struct{})
+	close(stop)
+
+	done := make(chan struct{})
+	go func() {
+		r.Maintain(stop)
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Maintain ignored an already-closed stop channel")
+	}
+	assert.NoFileExists(t, r.Path(), "a stopped maintainer must not arm")
+}
+
+func TestDefaultStoreReserveInterval_IsBounded(t *testing.T) {
+	assert.Positive(t, svcnats.DefaultStoreReserveInterval)
+	assert.LessOrEqual(t, svcnats.DefaultStoreReserveInterval, time.Hour,
+		"a node that spent its headroom must re-arm well inside a working day")
+}
+
+func TestDefaultStoreReserveBytes_LeavesRecoveryHeadroom(t *testing.T) {
+	assert.Positive(t, svcnats.DefaultStoreReserveBytes)
+	assert.Less(t, svcnats.DefaultStoreReserveBytes, int64(1)<<30,
+		"the reserve is held on every node and must stay small next to max_file_store")
+}