diff --git a/spinifex/services/nats/nats.go b/spinifex/services/nats/nats.go index 683621c22..14b17e3a1 100644 --- a/spinifex/services/nats/nats.go +++ b/spinifex/services/nats/nats.go @@ -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 diff --git a/spinifex/services/nats/reserve.go b/spinifex/services/nats/reserve.go new file mode 100644 index 000000000..317e9d5b4 --- /dev/null +++ b/spinifex/services/nats/reserve.go @@ -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 +} diff --git a/spinifex/services/nats/reserve_test.go b/spinifex/services/nats/reserve_test.go new file mode 100644 index 000000000..3f0620e61 --- /dev/null +++ b/spinifex/services/nats/reserve_test.go @@ -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") +}