snapshots: never of a save a sync is writing, never incomplete

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.

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

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 —

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 == "" {

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)

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)

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) {

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.