From df1f134e0959b8cdb0dd2df0d2564e67bf8ab5ce Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 9 Jul 2026 16:37:58 +0800 Subject: [PATCH 1/3] perf(pool): move the claims-journal write off the manager mutex claimStore.save marshaled and wrote+renamed the whole claim map while the caller held m.mu, so every claim/release/wake/hibernate/reap stalled any concurrent data-plane m.mu acquire for the syscall duration (measured 1.2-1.9ms at 100-1000 live claims). Split it: snapshot() marshals under m.mu and stamps a sequence; commit() writes off the mutex, serialized and coalescing (an older sequence never overwrites a newer one). The six hot sites marshal-under-lock, unlock, commit; rollback paths re-snapshot and re-commit so disk converges to the rolled-back set. --- sandboxd/pool/archive.go | 40 +++++++++++++++------------ sandboxd/pool/claim.go | 47 +++++++++++++++++++------------- sandboxd/pool/claims.go | 53 +++++++++++++++++++++++++++++++++--- sandboxd/pool/claims_test.go | 37 +++++++++++++++++++++++++ sandboxd/pool/hibernate.go | 10 ++++--- 5 files changed, 143 insertions(+), 44 deletions(-) diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index 34295eb..f4b2277 100644 --- a/sandboxd/pool/archive.go +++ b/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.store.commit(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.store.commit(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. diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index 9e82249..6ff4f8f 100644 --- a/sandboxd/pool/claim.go +++ b/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.store.commit(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.store.commit(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 { + js = m.store.snapshot(m.claimed) + } + m.mu.Unlock() + + logger := log.WithFunc("pool.reapOnce") if len(expired) > 0 { - if saveErr = m.store.save(m.claimed); saveErr != nil { + 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.store.commit(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) { diff --git a/sandboxd/pool/claims.go b/sandboxd/pool/claims.go index 391ab4d..11bfcb5 100644 --- a/sandboxd/pool/claims.go +++ b/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,42 @@ 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, holding no lock itself — for callers +// not already under the manager mutex (Reconcile). Hot-path callers split +// snapshot()/commit() around the mutex instead. +func (s *claimStore) save(claims map[string]*types.Sandbox) error { + return s.commit(s.snapshot(claims)) +} diff --git a/sandboxd/pool/claims_test.go b/sandboxd/pool/claims_test.go index 67f090b..67295af 100644 --- a/sandboxd/pool/claims_test.go +++ b/sandboxd/pool/claims_test.go @@ -4,6 +4,7 @@ import ( "maps" "os" "path/filepath" + "sync" "testing" "time" @@ -49,3 +50,39 @@ func TestStoreLoadCorruptFile(t *testing.T) { t.Error("load succeeded on corrupt file") } } + +// TestClaimStoreCommitCoalesces asserts an older snapshot never overwrites a +// newer one already on disk — the guard that lets the write leave the manager +// mutex without losing the latest state. +func TestClaimStoreCommitCoalesces(t *testing.T) { + s := newClaimStore(t.TempDir()) + one := map[string]*types.Sandbox{"sb_a": {ID: "sb_a"}} + two := map[string]*types.Sandbox{"sb_a": {ID: "sb_a"}, "sb_b": {ID: "sb_b"}} + + older := s.snapshot(one) // seq 1 + newer := s.snapshot(two) // seq 2 + if err := s.commit(newer); err != nil { + t.Fatalf("commit newer: %v", err) + } + if err := s.commit(older); err != nil { // must be a no-op, not an overwrite + t.Fatalf("commit older: %v", err) + } + if got, err := s.load(); err != nil || len(got) != 2 { + t.Errorf("stale snapshot overwrote the newer state: %d claims (%v), want 2", len(got), err) + } +} + +// TestClaimStoreConcurrentCommits exercises the writer off the manager mutex: +// many snapshot+commit pairs race, and the file must stay parseable (-race). +func TestClaimStoreConcurrentCommits(t *testing.T) { + s := newClaimStore(t.TempDir()) + claims := map[string]*types.Sandbox{"sb_a": {ID: "sb_a"}} + var wg sync.WaitGroup + for range 50 { + wg.Go(func() { _ = s.commit(s.snapshot(claims)) }) + } + wg.Wait() + if got, err := s.load(); err != nil || len(got) != 1 { + t.Fatalf("after concurrent commits: %+v, %v", got, err) + } +} diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go index e3bdd6f..097feb1 100644 --- a/sandboxd/pool/hibernate.go +++ b/sandboxd/pool/hibernate.go @@ -170,15 +170,17 @@ func (m *Manager) idleHibernate(ctx context.Context, id, token string, sweepStar func (m *Manager) commitTransition(ctx context.Context, sb *types.Sandbox, snap, sock string) bool { m.mu.Lock() live := m.claimed[sb.ID] == sb - var saveErr error + var js claimSnapshot if live { sb.HibernateSnap = snap sb.VsockSocket = sock - saveErr = m.store.save(m.claimed) + js = m.store.snapshot(m.claimed) } m.mu.Unlock() - if saveErr != nil { - log.WithFunc("pool.commitTransition").Warnf(ctx, "persist claims: %v", saveErr) + if live { + if err := m.store.commit(js); err != nil { + log.WithFunc("pool.commitTransition").Warnf(ctx, "persist claims: %v", err) + } } return live } From 62635df03b2f7f33aaa59c0cbca143d622abaf9d Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 9 Jul 2026 17:00:35 +0800 Subject: [PATCH 2/3] review(pool): retry the rollback re-commit; fix Reconcile save doc Adversarial review (an agent + codex) found the six-site write split sound (coalescing, lock order, seq/written races all verified) with one real edge: a rolled-back snapshot's best-effort re-commit was discarded, so a double I/O fault with a concurrent successful commit between them could leave disk stuck disagreeing with memory until an unrelated later write. Retry the re-commit so a transient fault converges and a coalesced result counts as done; a persistent failure warns (memory authoritative + Reconcile on start). Also correct save()'s doc: Reconcile holds m.mu (startup, uncontended). --- sandboxd/pool/archive.go | 4 ++-- sandboxd/pool/claim.go | 21 ++++++++++++++++++--- sandboxd/pool/claims.go | 7 ++++--- sandboxd/pool/pool.go | 2 ++ 4 files changed, 26 insertions(+), 8 deletions(-) diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index f4b2277..ee9b347 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -127,7 +127,7 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error { rb := m.store.snapshot(m.claimed) m.mu.Unlock() sb.Transition.Unlock() - _ = m.store.commit(rb) + m.recommit(ctx, rb) m.deleteOrphanArchiveCk(ctx, ck.ID) return fmt.Errorf("archive %s: persist claims: %w", sb.ID, saveErr) } @@ -196,7 +196,7 @@ func (m *Manager) commitWake(ctx context.Context, sb *types.Sandbox, vmName, soc sb.VMName, sb.VsockSocket, sb.ArchiveCk, sb.Deadline = "", "", ck, deadline rb := m.store.snapshot(m.claimed) m.mu.Unlock() - _ = m.store.commit(rb) + m.recommit(ctx, rb) log.WithFunc("pool.commitWake").Warnf(ctx, "persist claims: %v", err) return true, false } diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index 6ff4f8f..c102cf4 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -105,7 +105,7 @@ func (m *Manager) Release(ctx context.Context, id, token string) error { m.tenantDelta(sb.Tenant, 1) rb := m.store.snapshot(m.claimed) m.mu.Unlock() - _ = m.store.commit(rb) + m.recommit(ctx, rb) log.WithFunc("pool.Release").Errorf(ctx, saveErr, "persist release of %s", id) return fmt.Errorf("release %s: %w", id, saveErr) } @@ -253,7 +253,7 @@ func (m *Manager) finalizeBatch(ctx context.Context, sbs []*types.Sandbox, ttl t } rb := m.store.snapshot(m.claimed) m.mu.Unlock() - _ = m.store.commit(rb) // converge disk to the rolled-back set + m.recommit(ctx, rb) // converge disk to the rolled-back set for _, sb := range sbs { m.destroy(ctx, sb.VMName) } @@ -318,7 +318,7 @@ func (m *Manager) reapOnce(ctx context.Context) { } rb := m.store.snapshot(m.claimed) m.mu.Unlock() - _ = m.store.commit(rb) + m.recommit(ctx, rb) logger.Error(ctx, saveErr, "persist reap; rolled back") return } @@ -358,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() diff --git a/sandboxd/pool/claims.go b/sandboxd/pool/claims.go index 11bfcb5..b9c226a 100644 --- a/sandboxd/pool/claims.go +++ b/sandboxd/pool/claims.go @@ -90,9 +90,10 @@ func (s *claimStore) commit(snap claimSnapshot) error { return nil } -// save marshals and commits in one call, holding no lock itself — for callers -// not already under the manager mutex (Reconcile). Hot-path callers split -// snapshot()/commit() around the mutex instead. +// 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)) } diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 9232839..3fadd85 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -39,6 +39,8 @@ const ( maxConcurrentRefills = 4 defaultTTL = 5 * time.Minute maxTTL = 24 * time.Hour + recommitAttempts = 3 + recommitBackoff = 20 * time.Millisecond // defaultMaxFork is the fork ceiling when a Manager is built from a Config // that skipped config.Load's defaulting (direct construction in tests). defaultMaxFork = 16 From 4fe477a5a70807bd6f9dedfa96992564be3e5984 Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 9 Jul 2026 18:46:06 +0800 Subject: [PATCH 3/3] test(pool): contention benchmark for the journal write split BenchmarkStorePersistContention measures a concurrent manager-mutex acquire's wait during a persist, under-lock (save) vs off-lock (snapshot+commit). Bare metal .79: 38882->227 ns/acquire @100 claims (171x), 53926->796 @1000 (68x). --- sandboxd/pool/claims_bench_test.go | 87 +++++++++++++++++++++++++----- 1 file changed, 74 insertions(+), 13 deletions(-) diff --git a/sandboxd/pool/claims_bench_test.go b/sandboxd/pool/claims_bench_test.go index 0f42bc0..71fc248 100644 --- a/sandboxd/pool/claims_bench_test.go +++ b/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") + } +}