sync: a game being played is held back from syncing until the session ends

internal/api/routes_peers.go

@@ -243,13 +243,15 @@ func (s *Server) handleProbePeer(w http.ResponseWriter, r *http.Request) {
 }
 
 // syncReason names why a sync did not happen, for a client to act on rather
-// than parse the message: "paused" (this device paused syncing), "held" (the
+// than parse the message: "paused" (this device paused syncing), "playing" (the game is being played here; it syncs when the session ends), "held" (the
 // game's save was emptied here and waits for an answer), "offline" (no other
 // device answered), or "error" for anything else.
 func syncReason(err error) string {
 	switch {
 	case errors.Is(err, syncengine.ErrPaused):
 		return "paused"
+	case errors.Is(err, syncengine.ErrPlayingHere):
+		return "playing"
 	case errors.Is(err, syncengine.ErrHeld):
 		return "held"
 	case errors.Is(err, p2p.ErrNoPeersOnline):

internal/daemon/daemon.go

@@ -186,6 +186,9 @@ func New(opts Options) (*Daemon, error) {
 
 	// A game's device settings are left out of syncing (devicesettings.go).
 	d.P2P.Sync.DeviceSettingsFor = d.deviceSettingsFor
+	// A game being played here is synced when its session ends
+	// (syncengine/playing.go).
+	d.P2P.Sync.Playing = func(gameID string) bool { return !d.PlayingSince(gameID).IsZero() }
 
 	// A restore or a branch switch rewrites a save folder; nothing syncing it
 	// may read it half-way (syncengine/settle.go).
@@ -273,6 +276,7 @@ func New(opts Options) (*Daemon, error) {
 		SyncWritingNow: func(gameID string) bool {
 			return d.P2P != nil && d.P2P.Sync != nil && d.P2P.Sync.WritingNow(gameID)
 		},
+		HoldSnapshot: d.holdSnapshotWhilePlaying,
 		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/daemon/sessions.go

@@ -42,6 +42,19 @@ type sessionState struct {
 	list func() ([]sessions.Proc, error)
 
 	mu sync.Mutex
+	// cpu: each process's processor time at the last poll, and when it last
+	// used any (activeProcesses).
+	cpu map[int]procSeen
+	// saveChanged: when each playing game's save last changed; quiet: where
+	// each one's quiet stretch began; quietEnded: games whose session was
+	// ended as quiet, by the save as it was then, while their process is still
+	// there (endQuietSessions).
+	saveChanged map[string]time.Time
+	quiet       map[string]quietWindow
+	quietEnded  map[string]string
+	// checkpoint is when each playing game last had a snapshot taken while
+	// it was played (holdSnapshotWhilePlaying).
+	checkpoint map[string]time.Time
 	// startHash is each playing game's save as it was when play began, to
 	// tell afterwards whether the session changed it.
 	startHash map[string]string
@@ -79,19 +92,23 @@ func (d *Daemon) PlayingSince(gameID string) time.Time {
 }
 
 // PollSessions looks at what is running once.
-func (d *Daemon) PollSessions() {
+func (d *Daemon) PollSessions() { d.pollSessionsAt(time.Now()) }
+
+func (d *Daemon) pollSessionsAt(now time.Time) {
 	targets, learn, err := d.sessionTargets()
 	if err != nil || len(targets) == 0 {
-		d.sessions.tracker.Poll(nil, time.Now())
+		d.sessions.tracker.Poll(nil, now)
 		return
 	}
 	procs, err := d.sessions.list()
 	if err != nil {
 		return
 	}
+	procs = d.activeProcessesAt(procs, now)
 	running := sessions.Running(procs, targets)
 	d.notePrograms(procs, targets, learn, running)
-	d.sessions.tracker.Poll(running, time.Now())
+	d.endQuietSessions(running, sessions.CPUOf(procs, targets), now)
+	d.sessions.tracker.Poll(running, now)
 }
 
 func (d *Daemon) runSessions(ctx context.Context) {
@@ -365,6 +382,8 @@ func (d *Daemon) sessionEnded(gameID string, started, ended time.Time) {
 	d.sessions.mu.Lock()
 	startHash := d.sessions.startHash[gameID]
 	delete(d.sessions.startHash, gameID)
+	delete(d.sessions.checkpoint, gameID)
+	delete(d.sessions.saveChanged, gameID)
 	seen := d.sessions.seen[gameID]
 	delete(d.sessions.seen, gameID)
 	d.sessions.mu.Unlock()
@@ -455,3 +474,211 @@ func (d *Daemon) endOpenSessions() {
 		}
 	}
 }
+
+// checkpointEvery is how often a game being played gets a snapshot of its
+// save: one per autosave pushed the snapshots worth keeping out, and none
+// until the session ends leaves nothing behind a crash.
+const checkpointEvery = 30 * time.Minute
+
+// holdSnapshotWhilePlaying tells the watcher not to snapshot a change to a
+// game being played, unless checkpointEvery has passed since play began or
+// since the last checkpoint. The session's end keeps the save as it was left
+// (sessionEnded).
+func (d *Daemon) holdSnapshotWhilePlaying(gameID string) bool {
+	since := d.PlayingSince(gameID)
+	if since.IsZero() {
+		return false
+	}
+	d.sessions.mu.Lock()
+	defer d.sessions.mu.Unlock()
+	if d.sessions.checkpoint == nil {
+		d.sessions.checkpoint = map[string]time.Time{}
+	}
+	now := time.Now()
+	if d.sessions.saveChanged == nil {
+		d.sessions.saveChanged = map[string]time.Time{}
+	}
+	d.sessions.saveChanged[gameID] = now // the game is saving: being played (endQuietSessions)
+	if !checkpointDue(since, d.sessions.checkpoint[gameID], now) {
+		return true
+	}
+	d.sessions.checkpoint[gameID] = now
+	return false
+}
+
+// checkpointDue says whether checkpointEvery has passed since play began, or
+// since the last checkpoint when there was one.
+func checkpointDue(since, last, now time.Time) bool {
+	if last.Before(since) {
+		last = since
+	}
+	return now.Sub(last) >= checkpointEvery
+}
+
+// activeProcesses leaves out the processes doing nothing. A game leaves one
+// behind when it closes or is killed — a crash handler, a stub of the same
+// name — and it kept a session open for as long as it lingered: fifteen
+// minutes after one game had closed. While a session is open the game's save
+// is held back from syncing (syncengine/playing.go), so this is what lets it
+// go once the game is gone.
+func (d *Daemon) activeProcesses(procs []sessions.Proc) []sessions.Proc {
+	return d.activeProcessesAt(procs, time.Now())
+}
+
+// procSeen is a process's processor time at the last poll, and when it was
+// last seen using any.
+type procSeen struct {
+	cpu  time.Duration
+	busy time.Time
+}
+
+// idleFor is how long a process may use no processor time at all before it
+// is no longer the game, however much or little memory it holds. A game
+// being played uses some all the time — a paused one still draws its menu —
+// but one minimised in the background can throttle to almost nothing, or be
+// suspended, and Windows trims its memory too. Ending its session then would
+// sync the save under it, which is what the session is there to stop (asked
+// in review on #117). Five minutes still lets go of what a closed game left
+// behind well before the quarter of an hour one lingered.
+const idleFor = 5 * time.Minute
+
+// busyCPU is the processor time between two polls that counts as using any.
+const busyCPU = 20 * time.Millisecond
+
+func (d *Daemon) activeProcessesAt(procs []sessions.Proc, now time.Time) []sessions.Proc {
+	d.sessions.mu.Lock()
+	defer d.sessions.mu.Unlock()
+	prev := d.sessions.cpu
+	next := make(map[int]procSeen, len(procs))
+	out := procs[:0:0]
+	for _, p := range procs {
+		before, seen := prev[p.PID]
+		if !p.Measured {
+			out = append(out, p) // cannot be told: counted
+			continue
+		}
+		s := procSeen{cpu: p.CPU, busy: before.busy}
+		if !seen || p.CPU-before.cpu >= busyCPU {
+			s.busy = now // new, or working
+		}
+		next[p.PID] = s
+		if now.Sub(s.busy) > idleFor {
+			continue // doing nothing for long enough not to be the game
+		}
+		out = append(out, p)
+	}
+	d.sessions.cpu = next
+	return out
+}
+
+// quietSessionLimit is how long a session may go without its save changing,
+// and its game's processes all but idle, before it is taken as over.
+//
+// While a game is played its save is held back from syncing
+// (syncengine/playing.go), so a session that never ended would hold the save
+// for good. A process left behind that still ticks over now and then — a
+// crash reporter, a launcher's helper — counts as running however long it
+// lingers: activeProcesses only lets go of one doing nothing at all.
+//
+// A game being played may well not save for an hour — one saved by hand, at
+// save points — but it keeps drawing, and uses far more processor time than
+// quietCPUShare. What a game left behind uses less. Ended, the session is kept
+// and synced as when the game closes.
+//
+// Never a session begun from outside (Tracker.Marked: the Deck plugin, `opensave
+// wrap`): what began it says when it ends, and knows.
+const quietSessionLimit = time.Hour
+
+// quietCPUShare is the share of one processor core, at most, that a game's
+// processes may have used over quietSessionLimit for its session to be taken
+// as over: 1%, some 36 seconds in an hour.
+const quietCPUShare = 100 // as a divisor
+
+// quietWindow is where a game's quiet stretch began: the processor time its
+// processes had used by then, and when.
+type quietWindow struct {
+	cpu time.Duration
+	at  time.Time
+}
+
+// endQuietSessions ends the sessions whose save has not changed, and whose
+// processes have all but idled, for quietSessionLimit; and keeps them from
+// starting again at once. running is the games seen running this poll, and
+// cpu their processes' processor time (sessions.CPUOf); a game ended this way
+// is taken out of running while its save is still as it was then. Called
+// before the tracker polls.
+//
+// Measured in windows of quietSessionLimit from the session's start or its
+// save's last change: at the end of one, the session is over if the game was
+// all but idle in it; otherwise a new window begins.
+func (d *Daemon) endQuietSessions(running map[string]int, cpu map[string]sessions.GameCPU, now time.Time) {
+	playing := d.sessions.tracker.Playing()
+	lastHash := func(gameID string) string {
+		if g, err := d.Store.GetGame(gameID); err == nil {
+			return g.LastManifestHash
+		}
+		return ""
+	}
+
+	var end []string
+	d.sessions.mu.Lock()
+	if d.sessions.quietEnded == nil {
+		d.sessions.quietEnded = map[string]string{}
+	}
+	if d.sessions.quiet == nil {
+		d.sessions.quiet = map[string]quietWindow{}
+	}
+	for gameID, hash := range d.sessions.quietEnded {
+		if _, still := running[gameID]; !still {
+			delete(d.sessions.quietEnded, gameID) // gone: its next start is a session again
+		} else if lastHash(gameID) != hash {
+			delete(d.sessions.quietEnded, gameID) // saved since: being played after all
+		} else {
+			delete(running, gameID)
+		}
+	}
+	for gameID := range d.sessions.quiet {
+		if _, ok := playing[gameID]; !ok {
+			delete(d.sessions.quiet, gameID)
+		}
+	}
+	for gameID, since := range playing {
+		g, seen := cpu[gameID]
+		if d.sessions.tracker.Marked(gameID) || !seen || !g.Measured {
+			delete(d.sessions.quiet, gameID) // begun from outside, or cannot be told
+			continue
+		}
+		w, has := d.sessions.quiet[gameID]
+		if !has {
+			d.sessions.quiet[gameID] = quietWindow{cpu: g.CPU, at: since} // the first window: from the start
+			continue
+		}
+		if d.sessions.saveChanged[gameID].After(w.at) || g.CPU < w.cpu {
+			d.sessions.quiet[gameID] = quietWindow{cpu: g.CPU, at: now} // saved, or a new process: a new window
+			continue
+		}
+		span := now.Sub(w.at)
+		if span < quietSessionLimit {
+			continue
+		}
+		if g.CPU-w.cpu >= span/quietCPUShare {
+			d.sessions.quiet[gameID] = quietWindow{cpu: g.CPU, at: now} // busy: being played
+			continue
+		}
+		delete(d.sessions.quiet, gameID)
+		d.sessions.quietEnded[gameID] = lastHash(gameID)
+		delete(running, gameID)
+		end = append(end, gameID)
+	}
+	d.sessions.mu.Unlock()
+
+	for _, gameID := range end {
+		name := gameID
+		if g, err := d.Store.GetGame(gameID); err == nil {
+			name = g.Name
+		}
+		d.Log.Log("info", fmt.Sprintf("%q has neither saved nor done anything for %s while seen running: taken as no longer played, so its save is kept and synced",
+			name, quietSessionLimit))
+		d.sessions.tracker.Finish(gameID, now)
+	}
+}