From 1c40ee3a65b4425f53676aca0c09a721984103a7 Mon Sep 17 00:00:00 2001 From: ChakraFusion <112588066+ChakraFusion@users.noreply.github.com> Date: Fri, 2 Oct 2026 15:44:43 +0200 Subject: [PATCH] snapshots: never of a save a sync is writing, never incomplete A pull of a save of many files overflowed the watcher's event queue again and again, and each overflow was taken for the game saving: an automatic snapshot of a half-pulled save every few seconds - 76 on one device for one pull - which pushed the complete ones out of retention. - The watcher asks whether a sync is writing the game (BeingWritten). If so, an overflow is taken for the sync's, and the burst is looked at only once the sync has finished. - A snapshot that could not read every file is discarded, not kept: it would stand in the history as a save that never existed. The watcher leaves the change unrecorded and takes it again at its next pass; a sync that needed the copy before replacing files does not replace them. - OpenSave's own temporary files (*.opensave.tmp) are never archived. Builds on the own-writes change (Liquid-co/OpenSave#23). Co-Authored-By: Claude Opus 5.5 --- internal/daemon/daemon.go | 6 ++++++ internal/p2p/syncengine/engine.go | 2 ++ internal/p2p/syncengine/settle.go | 34 +++++++++++++++++++++++++++++ internal/snapshot/snapshot.go | 22 ++++++++++++++++--- internal/snapshot/zip.go | 5 +++++ internal/snapshot/zip_roots.go | 4 ++++ internal/watcher/owntouch_test.go | 30 ++++++++++++++++++++++++++ internal/watcher/watcher.go | 36 ++++++++++++++++++++++++++++++- 8 files changed, 135 insertions(+), 4 deletions(-) diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index 3d732f86..331c296b 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -236,6 +236,12 @@ func New(opts Options) (*Daemon, error) { _, err := snaps.Create(gameID, "", true) return err }, + SyncWriting: func(gameID string) bool { + return d.P2P != nil && d.P2P.Sync != nil && d.P2P.Sync.BeingWritten(gameID) + }, + SyncWritingNow: func(gameID string) bool { + return d.P2P != nil && d.P2P.Sync != nil && d.P2P.Sync.WritingNow(gameID) + }, OnChanged: func(gameID string) { // An emptied save is noticed as it happens, and said on screen, // even with no other device online to hold it back from. diff --git a/internal/p2p/syncengine/engine.go b/internal/p2p/syncengine/engine.go index 2e6b7ce6..2e26e227 100644 --- a/internal/p2p/syncengine/engine.go +++ b/internal/p2p/syncengine/engine.go @@ -127,6 +127,8 @@ type Engine struct { // on this device right now (settle.go). settleMu sync.Mutex gates map[string]*saveGate + // writtenAt is when a sync last stopped writing each game (BeingWritten). + writtenAt map[string]time.Time // servedMu guards served: the save states this device recently handed // each peer, per game (served.go). servedMu sync.Mutex diff --git a/internal/p2p/syncengine/settle.go b/internal/p2p/syncengine/settle.go index 34942e10..a66c1c8d 100644 --- a/internal/p2p/syncengine/settle.go +++ b/internal/p2p/syncengine/settle.go @@ -131,10 +131,44 @@ func (e *Engine) Writing(gameID string) (done func()) { once = true g := e.gateLocked(gameID) g.writers-- + if e.writtenAt == nil { + e.writtenAt = map[string]time.Time{} + } + e.writtenAt[gameID] = time.Now() e.wakeLocked(gameID, g) } } +// writeEcho is how long after a sync stops writing a game its file events may +// still be arriving. +const writeEcho = 10 * time.Second + +// BeingWritten reports whether a sync on this device is writing a game's save +// now, or was a moment ago — what the watcher asks before taking a burst it +// could not attribute (an overflow of file events) for a change of the game's. +// A pull of a save of many files overflows the event queue again and again, +// and each one read as the game saving: an automatic snapshot of a half-pulled +// save every few seconds, which pushed the real ones out of retention. +func (e *Engine) BeingWritten(gameID string) bool { + if e.WritingNow(gameID) { + return true + } + e.settleMu.Lock() + defer e.settleMu.Unlock() + return time.Since(e.writtenAt[gameID]) < writeEcho +} + +// WritingNow reports whether a sync on this device is writing a game's save at +// this moment — what the watcher waits out before looking at a burst, so it +// never judges a save half-way between two states. Not the moments after: +// a change the game makes then is the game's, and its snapshot is not held up. +func (e *Engine) WritingNow(gameID string) bool { + e.settleMu.Lock() + defer e.settleMu.Unlock() + g := e.gates[gameID] + return g != nil && (g.writers > 0 || g.waitingWriters > 0) +} + // Reading holds the game's save still for a sync to read it: it waits until // nothing is writing it, then keeps writers out until done is called. It // returns ErrSettling if ctx ends first. Hold it only for the reading itself — diff --git a/internal/snapshot/snapshot.go b/internal/snapshot/snapshot.go index f483bde4..f595fd90 100644 --- a/internal/snapshot/snapshot.go +++ b/internal/snapshot/snapshot.go @@ -41,6 +41,18 @@ import ( // before starting one. type UploadHook func(zipPath, remoteFileName string) +// IncompleteError is a snapshot discarded because files of the save could not +// be read — in use by the game, or changing while it was archived. +type IncompleteError struct { + Game string + Skipped []string +} + +func (e *IncompleteError) Error() string { + return fmt.Sprintf("snapshot of %q discarded: %d file(s) could not be read (e.g. %s) — taken again at the next chance", + e.Game, len(e.Skipped), e.Skipped[0]) +} + // Manager performs snapshot/branch operations against the store and // filesystem. type Manager struct { @@ -249,9 +261,13 @@ func (m *Manager) createOnBranchFrom(gameID, branch, from, comment string, isSys return store.Snapshot{}, fmt.Errorf("zip save data: %w", err) } defer os.Remove(stagingPath) // no-op once renamed - if len(skipped) > 0 && m.Log != nil { - sample := skipped[0] - m.Log("warn", fmt.Sprintf("snapshot of %q skipped %d unreadable file(s), e.g. %s", game.Name, len(skipped), sample)) + // A snapshot missing files is not kept: it would stand in the history as + // a save that never existed — and, counted against retention, push out a + // complete one. The caller tries again at the next chance (the watcher + // leaves the change unrecorded, so its next pass takes it again; a sync + // that needed the copy before replacing files does not replace them). + if len(skipped) > 0 { + return store.Snapshot{}, &IncompleteError{Game: game.Name, Skipped: skipped} } if comment == "" { diff --git a/internal/snapshot/zip.go b/internal/snapshot/zip.go index 803c60a8..804f7cf8 100644 --- a/internal/snapshot/zip.go +++ b/internal/snapshot/zip.go @@ -95,6 +95,11 @@ func zipPathCapturing(sourcePath, outPath string, reuse *reuseSource) (skipped [ } return nil } + // OpenSave's own temporary file for a file being pulled: never part + // of the save, and gone by the time it is read. + if strings.HasSuffix(rel, delta.TmpSuffix) { + return nil + } info, infoErr := d.Info() if infoErr != nil { skipped = append(skipped, path) diff --git a/internal/snapshot/zip_roots.go b/internal/snapshot/zip_roots.go index c3618b6c..fac78559 100644 --- a/internal/snapshot/zip_roots.go +++ b/internal/snapshot/zip_roots.go @@ -144,6 +144,10 @@ func archiveInto(w *zip.Writer, sourcePath, prefix, root string, reuse *reuseSou _, err := w.CreateHeader(&zip.FileHeader{Name: prefix + rel + "/", Method: zip.Store}) return err } + // OpenSave's own temporary file for a file being pulled (zip.go). + if strings.HasSuffix(rel, delta.TmpSuffix) { + return nil + } info, infoErr := d.Info() if infoErr != nil { skipped = append(skipped, path) diff --git a/internal/watcher/owntouch_test.go b/internal/watcher/owntouch_test.go index 6a1e8134..d984bc29 100644 --- a/internal/watcher/owntouch_test.go +++ b/internal/watcher/owntouch_test.go @@ -4,6 +4,7 @@ import ( "fmt" "os" "path/filepath" + "sync/atomic" "testing" "time" @@ -103,6 +104,35 @@ func TestOwnNewFolders_AreNotSnapshotted(t *testing.T) { } } +// While a sync is writing the game, the watcher looks at what changed only +// once it has finished — never at a save half-way between two states — and a +// change of the game's is still snapshotted then. +func TestWhileASyncWrites_TheWatcherWaits(t *testing.T) { + saveDir := t.TempDir() + col := newCollector() + var writing atomic.Bool + cb := col.callbacks() + cb.SyncWriting = func(string) bool { return writing.Load() } + cb.SyncWritingNow = cb.SyncWriting + eng := New(cb) + defer eng.Stop() + if err := eng.Watch("game1", saveDir); err != nil { + t.Fatal(err) + } + writing.Store(true) + if err := os.WriteFile(filepath.Join(saveDir, "slot1.sav"), []byte("played"), 0o666); err != nil { + t.Fatal(err) + } + time.Sleep(5 * time.Second) + if n := col.snapshotCount(); n != 0 { + t.Fatalf("%d snapshot(s) while a sync was writing the game", n) + } + writing.Store(false) + if !waitFor(t, 10*time.Second, func() bool { return col.snapshotCount() >= 1 }) { + t.Fatal("the game's change was not snapshotted once the sync had finished") + } +} + // A burst that mixes OpenSave's own changes with one from the game is a real // change and is snapshotted. func TestMixedChanges_AreSnapshotted(t *testing.T) { diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go index 787c8a6b..e708b7d0 100644 --- a/internal/watcher/watcher.go +++ b/internal/watcher/watcher.go @@ -119,6 +119,25 @@ type Callbacks struct { OnChanged func(gameID string) // Log receives human-readable watcher activity. May be nil. Log func(level, msg string) + // SyncWriting reports whether a sync on this device is writing the game + // now, or was a moment ago: a burst the watcher cannot attribute — an + // overflow of file events — is then taken for the sync's. May be nil. + SyncWriting func(gameID string) bool + // SyncWritingNow reports whether a sync is writing the game at this + // moment: the watcher looks at the burst once it has finished. May be nil. + SyncWritingNow func(gameID string) bool +} + +// syncWriting reports whether a sync is or was just writing a game +// (Callbacks.SyncWriting). +func (e *Engine) syncWriting(gameID string) bool { + return e.cb.SyncWriting != nil && e.cb.SyncWriting(gameID) +} + +// syncWritingNow reports whether a sync is writing a game at this moment +// (Callbacks.SyncWritingNow). +func (e *Engine) syncWritingNow(gameID string) bool { + return e.cb.SyncWritingNow != nil && e.cb.SyncWritingNow(gameID) } // Engine owns one watch goroutine per tracked game. @@ -751,7 +770,16 @@ func (e *Engine) run(ctx context.Context, gw *gameWatch) { // directory and allocates no second buffer for one it already // has. gw.rewatch.Store(true) - gw.otherEvents = true // the missed events could be anyone's + // The missed events could be anyone's — unless a sync is + // writing this game, which is what overflows the queue: a + // pull of a save of many files does, again and again, and + // each was taken for the game saving, an automatic snapshot + // of a half-pulled save every few seconds. + if e.syncWriting(gw.gameID) { + gw.ownEvents = true + } else { + gw.otherEvents = true + } } else { e.log("warn", fmt.Sprintf("watching %q: %v", gw.gameID, err)) continue @@ -761,6 +789,12 @@ func (e *Engine) run(ctx context.Context, gw *gameWatch) { case <-debounceC: debounce = nil debounceC = nil + // A sync still writing this game: look at the burst once it has + // finished, not at a save half-way between two states. + if e.syncWritingNow(gw.gameID) { + resetDebounce() + continue + } // Re-register, so a folder that went unwatched is watched from // here on, and read the tree below, so it is found now. Doing only // one of those leaves it correct today and silent tomorrow.