perf(pool): move the claims-journal write off the manager mutex (F1)

sandboxd/pool/archive.go

@@ -116,18 +116,21 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error {
 	} else {
 		sb.Deadline = time.Time{} // keep forever: no retention (reap skips a zero deadline)
 	}
-	saveErr := m.store.save(m.claimed)
-	if saveErr != nil {
+	js := m.store.snapshot(m.claimed)
+	m.mu.Unlock()
+	if saveErr := m.store.commit(js); saveErr != nil {
 		// Roll back so memory matches the still-durable hibernated record; drop the orphan ck.
+		m.mu.Lock()
 		sb.ArchiveCk = ""
 		sb.HibernateSnap, sb.VsockSocket, sb.VMName = snap, sock, vmName
 		sb.Deadline = prevDeadline
+		rb := m.store.snapshot(m.claimed)
 		m.mu.Unlock()
 		sb.Transition.Unlock()
+		m.recommit(ctx, rb)
 		m.deleteOrphanArchiveCk(ctx, ck.ID)
 		return fmt.Errorf("archive %s: persist claims: %w", sb.ID, saveErr)
 	}
-	m.mu.Unlock()
 	sb.Transition.Unlock()
 	// Committed: the store ck is authoritative now; reclaim the local footprint.
 	if rmErr := m.eng.Remove(ctx, vmName); rmErr != nil {
@@ -179,23 +182,26 @@ func (m *Manager) wakeArchived(ctx context.Context, sb *types.Sandbox) (string,
 func (m *Manager) commitWake(ctx context.Context, sb *types.Sandbox, vmName, sock string) (live, saved bool) {
 	m.mu.Lock()
 	live = m.claimed[sb.ID] == sb
-	var saveErr error
-	if live {
-		ck, deadline := sb.ArchiveCk, sb.Deadline
-		sb.VMName, sb.VsockSocket, sb.ArchiveCk = vmName, sock, ""
-		sb.Deadline = time.Now().Add(clampTTL(0)) // a woken sandbox is a fresh lease
-		if saveErr = m.store.save(m.claimed); saveErr != nil {
-			sb.VMName, sb.VsockSocket, sb.ArchiveCk, sb.Deadline = "", "", ck, deadline
-		}
+	if !live {
+		m.mu.Unlock()
+		return false, false
 	}
+	ck, deadline := sb.ArchiveCk, sb.Deadline
+	sb.VMName, sb.VsockSocket, sb.ArchiveCk = vmName, sock, ""
+	sb.Deadline = time.Now().Add(clampTTL(0)) // a woken sandbox is a fresh lease
+	js := m.store.snapshot(m.claimed)
 	m.mu.Unlock()
-	if saveErr != nil {
-		log.WithFunc("pool.commitWake").Warnf(ctx, "persist claims: %v", saveErr)
-	}
-	if live && saveErr == nil {
-		sb.Touch() // restart the idle→hibernate→archive clock
+	if err := m.store.commit(js); err != nil {
+		m.mu.Lock()
+		sb.VMName, sb.VsockSocket, sb.ArchiveCk, sb.Deadline = "", "", ck, deadline
+		rb := m.store.snapshot(m.claimed)
+		m.mu.Unlock()
+		m.recommit(ctx, rb)
+		log.WithFunc("pool.commitWake").Warnf(ctx, "persist claims: %v", err)
+		return true, false
 	}
-	return live, saveErr == nil
+	sb.Touch() // restart the idle→hibernate→archive clock
+	return true, true
 }
 
 // deleteOrphanArchiveCk drops the published ck when archive() aborts pre-commit.

sandboxd/pool/claim.go

@@ -97,13 +97,15 @@ func (m *Manager) Release(ctx context.Context, id, token string) error {
 	delete(m.claimed, id)
 	m.tenantDelta(sb.Tenant, -1)
 	snap, ck, vmName := sb.HibernateSnap, sb.ArchiveCk, sb.VMName
-	saveErr := m.store.save(m.claimed)
-	if saveErr != nil {
+	js := m.store.snapshot(m.claimed)
+	m.mu.Unlock()
+	if saveErr := m.store.commit(js); saveErr != nil {
+		m.mu.Lock()
 		m.claimed[id] = sb // roll back so memory matches the still-durable claim; ck stays pinned
 		m.tenantDelta(sb.Tenant, 1)
-	}
-	m.mu.Unlock()
-	if saveErr != nil {
+		rb := m.store.snapshot(m.claimed)
+		m.mu.Unlock()
+		m.recommit(ctx, rb)
 		log.WithFunc("pool.Release").Errorf(ctx, saveErr, "persist release of %s", id)
 		return fmt.Errorf("release %s: %w", id, saveErr)
 	}
@@ -241,15 +243,17 @@ func (m *Manager) finalizeBatch(ctx context.Context, sbs []*types.Sandbox, ttl t
 		m.claimed[sb.ID] = sb
 		m.tenantDelta(sb.Tenant, 1)
 	}
-	saveErr := m.store.save(m.claimed)
-	if saveErr != nil {
+	js := m.store.snapshot(m.claimed)
+	m.mu.Unlock()
+	if saveErr := m.store.commit(js); saveErr != nil {
+		m.mu.Lock()
 		for _, sb := range sbs {
 			delete(m.claimed, sb.ID)
 			m.tenantDelta(sb.Tenant, -1)
 		}
-	}
-	m.mu.Unlock()
-	if saveErr != nil {
+		rb := m.store.snapshot(m.claimed)
+		m.mu.Unlock()
+		m.recommit(ctx, rb) // converge disk to the rolled-back set
 		for _, sb := range sbs {
 			m.destroy(ctx, sb.VMName)
 		}
@@ -294,9 +298,16 @@ func (m *Manager) reapOnce(ctx context.Context) {
 			m.tenantDelta(sb.Tenant, -1)
 		}
 	}
-	var saveErr error
+	var js claimSnapshot
 	if len(expired) > 0 {
-		if saveErr = m.store.save(m.claimed); saveErr != nil {
+		js = m.store.snapshot(m.claimed)
+	}
+	m.mu.Unlock()
+
+	logger := log.WithFunc("pool.reapOnce")
+	if len(expired) > 0 {
+		if saveErr := m.store.commit(js); saveErr != nil {
+			m.mu.Lock()
 			for _, v := range expired {
 				if v.action == reapArchive {
 					delete(m.archiving, v.id) // never removed from m.claimed
@@ -305,15 +316,13 @@ func (m *Manager) reapOnce(ctx context.Context) {
 					m.tenantDelta(v.tenant, 1)
 				}
 			}
+			rb := m.store.snapshot(m.claimed)
+			m.mu.Unlock()
+			m.recommit(ctx, rb)
+			logger.Error(ctx, saveErr, "persist reap; rolled back")
+			return
 		}
 	}
-	m.mu.Unlock()
-
-	logger := log.WithFunc("pool.reapOnce")
-	if saveErr != nil {
-		logger.Error(ctx, saveErr, "persist reap; rolled back")
-		return
-	}
 	// Engine subprocesses (worst case minutes on a hung stop): fan out bounded
 	// so a big batch never stalls the ticker loop.
 	m.runBounded(ctx, len(expired), func(ctx context.Context, i int) {
@@ -349,6 +358,21 @@ func (m *Manager) purgeArchiveCk(ctx context.Context, id, ck, tenant string) {
 	m.recordUsage(ctx, usageEvent{Event: "archive_delete", ID: id, Reference: ck, Tenant: tenant})
 }
 
+// recommit best-effort persists a rolled-back snapshot so disk converges to it
+// after a failed commit: a transient fault clears on retry, and a coalesced
+// result means a newer write already superseded this state. A persistent
+// failure only warns; memory is authoritative and Reconcile fixes it on start.
+func (m *Manager) recommit(ctx context.Context, snap claimSnapshot) {
+	var err error
+	for range recommitAttempts {
+		if err = m.store.commit(snap); err == nil {
+			return
+		}
+		time.Sleep(recommitBackoff)
+	}
+	log.WithFunc("pool.recommit").Errorf(ctx, err, "persist rolled-back claims after retries")
+}
+
 func (m *Manager) claim(id, token string) (*types.Sandbox, bool) {
 	m.mu.Lock()
 	defer m.mu.Unlock()

sandboxd/pool/claims.go

@@ -7,15 +7,35 @@ import (
 	"io/fs"
 	"os"
 	"path/filepath"
+	"sync"
+	"sync/atomic"
 
 	"github.com/cocoonstack/sandbox/sandboxd/types"
 )
 
+// claimSnapshot is a sequenced, pre-marshaled claim set awaiting a durable write.
+type claimSnapshot struct {
+	raw []byte
+	seq uint64
+	err error
+}
+
 // claimStore persists claimed sandboxes across daemon restarts. Warm pool
 // VMs are deliberately not persisted: they are cheap to rebuild and unsafe
 // to trust after an unsupervised gap.
+//
+// A write is split so the manager mutex covers only the marshal (a consistent
+// snapshot of the claim map), not the file syscalls: snapshot() runs under the
+// manager mutex, commit() writes off it. commit is serialized and coalescing —
+// a snapshot no newer than one already on disk is dropped, because sequence
+// order matches the manager-mutex mutation order, so the newest write wins and
+// an older one only ever describes a superseded state.
 type claimStore struct {
 	path string
+
+	seq     atomic.Uint64 // sequence source; bumped in snapshot (mutation order)
+	mu      sync.Mutex
+	written uint64 // highest sequence on disk; guarded by mu
 }
 
 func newClaimStore(dataDir string) *claimStore {
@@ -37,17 +57,43 @@ func (s *claimStore) load() (map[string]*types.Sandbox, error) {
 	return claims, nil
 }
 
-func (s *claimStore) save(claims map[string]*types.Sandbox) error {
+// snapshot marshals the claim set and stamps it with the next sequence. The
+// caller holds the manager mutex, so the map is consistent and sequences are
+// handed out in mutation order; the file write happens later in commit().
+func (s *claimStore) snapshot(claims map[string]*types.Sandbox) claimSnapshot {
+	seq := s.seq.Add(1)
 	raw, err := json.Marshal(claims)
-	if err != nil {
-		return fmt.Errorf("encode claims: %w", err)
+	return claimSnapshot{raw: raw, seq: seq, err: err}
+}
+
+// commit durably writes a snapshot off the manager mutex. Serialized by s.mu so
+// writes never interleave; coalescing so a snapshot no newer than the last
+// written is a no-op — a higher sequence already persisted a later state that
+// supersedes it, so the caller's intent is satisfied.
+func (s *claimStore) commit(snap claimSnapshot) error {
+	if snap.err != nil {
+		return fmt.Errorf("encode claims: %w", snap.err)
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if snap.seq <= s.written {
+		return nil
 	}
 	tmp := s.path + ".tmp"
-	if err := os.WriteFile(tmp, raw, 0o600); err != nil {
+	if err := os.WriteFile(tmp, snap.raw, 0o600); err != nil {
 		return fmt.Errorf("write claims: %w", err)
 	}
 	if err := os.Rename(tmp, s.path); err != nil {
 		return fmt.Errorf("commit claims: %w", err)
 	}
+	s.written = snap.seq
 	return nil
 }
+
+// save marshals and commits in one call — the combined form for the startup
+// Reconcile pass, which holds the manager mutex but runs before any claim or
+// housekeeping tick can contend. Hot-path callers split snapshot()/commit()
+// around the mutex so the write leaves it.
+func (s *claimStore) save(claims map[string]*types.Sandbox) error {
+	return s.commit(s.snapshot(claims))
+}

sandboxd/pool/claims_bench_test.go

@@ -2,29 +2,34 @@ package pool
 
 import (
 	"fmt"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/cocoonstack/sandbox/sandboxd/types"
 )
 
-// BenchmarkStoreSaveScaling measures the claims-journal write path (marshal
-// + write + rename, inside the pool mutex today) as the live-claim set
-// grows. M4-6 is trigger-gated: move save() to a store-owned sequential
-// writer only if this shows the in-mutex write dominating at scale.
+func benchClaims(n int) map[string]*types.Sandbox {
+	claims := make(map[string]*types.Sandbox, n)
+	for i := range n {
+		id := fmt.Sprintf("sb_%016x", i)
+		claims[id] = &types.Sandbox{
+			ID: id, Token: "tok", VMName: "sbx-" + id,
+			Key:      types.PoolKey{Template: "rt:24.04", Net: types.NetNone, Size: types.SizeSmall},
+			Deadline: time.Now().Add(time.Hour),
+		}
+	}
+	return claims
+}
+
+// BenchmarkStoreSaveScaling measures the claims-journal persist cost (marshal
+// + write + rename) as the live-claim set grows.
 func BenchmarkStoreSaveScaling(b *testing.B) {
 	for _, n := range []int{10, 100, 1000} {
 		b.Run(fmt.Sprintf("claims=%d", n), func(b *testing.B) {
 			st := newClaimStore(b.TempDir())
-			claims := make(map[string]*types.Sandbox, n)
-			for i := range n {
-				id := fmt.Sprintf("sb_%016x", i)
-				claims[id] = &types.Sandbox{
-					ID: id, Token: "tok", VMName: "sbx-" + id,
-					Key:      types.PoolKey{Template: "rt:24.04", Net: types.NetNone, Size: types.SizeSmall},
-					Deadline: time.Now().Add(time.Hour),
-				}
-			}
+			claims := benchClaims(n)
 			b.ResetTimer()
 			for range b.N {
 				if err := st.save(claims); err != nil {
@@ -34,3 +39,59 @@ func BenchmarkStoreSaveScaling(b *testing.B) {
 		})
 	}
 }
+
+// BenchmarkStorePersistContention reports how long a concurrent manager-mutex
+// acquire (a data-plane op) waits while the journal is being persisted, under
+// the old "write under the lock" path (save) vs the new "marshal under the lock,
+// write off it" split (snapshot+commit). The ns/acquire metric is the win.
+func BenchmarkStorePersistContention(b *testing.B) {
+	for _, n := range []int{100, 1000} {
+		claims := benchClaims(n)
+		b.Run(fmt.Sprintf("under-lock/n=%d", n), func(b *testing.B) {
+			benchPersistContention(b, claims, false)
+		})
+		b.Run(fmt.Sprintf("off-lock/n=%d", n), func(b *testing.B) {
+			benchPersistContention(b, claims, true)
+		})
+	}
+}
+
+func benchPersistContention(b *testing.B, claims map[string]*types.Sandbox, offLock bool) {
+	s := newClaimStore(b.TempDir())
+	var mu sync.Mutex
+	var waitNs, waits atomic.Int64
+	done := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-done:
+				return
+			default:
+			}
+			t := time.Now()
+			mu.Lock()
+			waitNs.Add(time.Since(t).Nanoseconds())
+			waits.Add(1)
+			mu.Unlock()
+		}
+	}()
+
+	b.ResetTimer()
+	for range b.N {
+		if offLock {
+			mu.Lock()
+			snap := s.snapshot(claims) // marshal under the lock
+			mu.Unlock()
+			_ = s.commit(snap) // write off the lock
+		} else {
+			mu.Lock()
+			_ = s.save(claims) // marshal + write both under the lock
+			mu.Unlock()
+		}
+	}
+	b.StopTimer()
+	close(done)
+	if w := waits.Load(); w > 0 {
+		b.ReportMetric(float64(waitNs.Load())/float64(w), "ns/acquire")
+	}
+}