diff --git a/internal/api/routes_peers.go b/internal/api/routes_peers.go index 4eb9b594..256cd05b 100644 --- a/internal/api/routes_peers.go +++ b/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): diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index 8abd9a26..bf1abdb3 100644 --- a/internal/daemon/daemon.go +++ b/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. diff --git a/internal/daemon/sessions.go b/internal/daemon/sessions.go index 67775f99..87dd1a65 100644 --- a/internal/daemon/sessions.go +++ b/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) + } +} diff --git a/internal/daemon/sessions_test.go b/internal/daemon/sessions_test.go index 362c27c7..b403721c 100644 --- a/internal/daemon/sessions_test.go +++ b/internal/daemon/sessions_test.go @@ -193,3 +193,188 @@ func TestSpokenLength(t *testing.T) { } } } + +// While a game is played, a change to its save is snapshotted once per +// checkpointEvery, not at every autosave; and it is not synced either way. +func TestSessionHoldsSnapshotsAndSyncWhilePlaying(t *testing.T) { + d := newTestDaemon(t) + g, _ := sessionGame(t, d, "autosaving") + if d.holdSnapshotWhilePlaying(g.ID) { + t.Fatal("held while nothing is being played") + } + d.sessions.list = func() ([]sessions.Proc, error) { return []sessions.Proc{{PID: 7, Exe: g.ExePath}}, nil } + d.PollSessions() + if d.PlayingSince(g.ID).IsZero() { + t.Fatal("setup: not playing") + } + if !d.holdSnapshotWhilePlaying(g.ID) { + t.Error("an autosave right after the game started was snapshotted") + } + if !d.P2P.Sync.PlayingHere(g.ID) { + t.Error("the sync engine does not know the game is being played") + } + // Checkpoints: half an hour after play began, then half an hour after + // the last one. + start := time.Now() + for _, c := range []struct { + last time.Time + after time.Duration + want bool + }{ + {time.Time{}, 10 * time.Minute, false}, + {time.Time{}, checkpointEvery, true}, + {start.Add(40 * time.Minute), 50 * time.Minute, false}, + {start.Add(40 * time.Minute), 70 * time.Minute, true}, + } { + if got := checkpointDue(start, c.last, start.Add(c.after)); got != c.want { + t.Errorf("checkpoint due %v after start (last %v) = %v, want %v", c.after, c.last, got, c.want) + } + } +} + +// A process doing nothing does not keep a session open, and with it the +// game's save held back from syncing. +func TestSessionEndsWhenOnlyAStubIsLeft(t *testing.T) { + d := newTestDaemon(t) + g, _ := sessionGame(t, d, "stub") + running := sessions.Proc{PID: 9, Exe: g.ExePath, Measured: true, CPU: time.Second} + d.sessions.list = func() ([]sessions.Proc, error) { return []sessions.Proc{running}, nil } + d.PollSessions() + if d.PlayingSince(g.ID).IsZero() { + t.Fatal("setup: the game was not seen running") + } + // The game closed; a stub of the same program lingers, doing nothing, and + // has for longer than idleFor. + stub := sessions.Proc{PID: 9, Exe: g.ExePath, Measured: true, CPU: time.Second} + d.sessions.mu.Lock() + d.sessions.cpu[9] = procSeen{cpu: time.Second, busy: time.Now().Add(-idleFor - time.Minute)} + d.sessions.mu.Unlock() + d.sessions.list = func() ([]sessions.Proc, error) { return []sessions.Proc{stub}, nil } + d.PollSessions() + since := d.PlayingSince(g.ID) + d.sessions.tracker.Finish(g.ID, since.Add(time.Hour)) // past the grace the tracker allows + d.PollSessions() + if !d.PlayingSince(g.ID).IsZero() { + t.Fatal("a lingering stub keeps the session open") + } + if d.holdSnapshotWhilePlaying(g.ID) { + t.Error("a change after the session ended was held") + } +} + +// A process that has used no processor time for idleFor is no longer the game, +// whatever it holds: what a killed game leaves behind. One quiet for less — +// a game minimised in the background, throttled or suspended — still is, so +// its session and the hold on its save go on. One working stays. +func TestActiveProcesses_QuietForLongIsNotTheGame(t *testing.T) { + d := newTestDaemon(t) + t0 := time.Now() + left := func(cpu time.Duration) sessions.Proc { + return sessions.Proc{PID: 1, Exe: `G:\Games\Crimson Desert\bin64\CrimsonDesert.exe`, Measured: true, CPU: cpu} + } + playing := func(cpu time.Duration) sessions.Proc { + return sessions.Proc{PID: 2, Exe: `G:\Games\Other\game.exe`, Measured: true, CPU: cpu} + } + count := func(ps []sessions.Proc) map[int]bool { + m := map[int]bool{} + for _, p := range ps { + m[p.PID] = true + } + return m + } + got := count(d.activeProcessesAt([]sessions.Proc{left(5 * time.Second), playing(time.Second)}, t0)) + if !got[1] || !got[2] { + t.Fatalf("first seen: %v, want both", got) + } + got = count(d.activeProcessesAt([]sessions.Proc{left(5 * time.Second), playing(2 * time.Second)}, t0.Add(30*time.Second))) + if !got[1] || !got[2] { + t.Errorf("half a minute in: %v, want both", got) + } + // Minimised and throttled: quiet for minutes, and still the game. + for _, at := range []time.Duration{70 * time.Second, 4 * time.Minute} { + got = count(d.activeProcessesAt([]sessions.Proc{left(5 * time.Second), playing(3 * time.Second)}, t0.Add(at))) + if !got[1] { + t.Errorf("quiet for %s: no longer counted, so a game in the background would lose its session", at) + } + } + got = count(d.activeProcessesAt([]sessions.Proc{left(5 * time.Second), playing(4 * time.Second)}, t0.Add(idleFor+10*time.Second))) + if got[1] { + t.Error("a process doing nothing for longer than idleFor still counts as the game") + } + if !got[2] { + t.Error("a game using the processor was dropped") + } +} + +// quietGame runs one game process whose processor time grows by perPoll at +// every poll. +func quietGame(t *testing.T, d *Daemon, id string, perPoll time.Duration) store.Game { + t.Helper() + g, _ := sessionGame(t, d, id) + cpu := time.Duration(0) + d.sessions.list = func() ([]sessions.Proc, error) { + cpu += perPoll + return []sessions.Proc{{PID: 7, Exe: g.ExePath, Measured: true, CPU: cpu}}, nil + } + return g +} + +// What a game left behind, still ticking over now and then, holds a session +// open, and with it the hold on the save. An hour without a save, all but +// idle, and the session is taken as over; the process still there does not +// start it again, and the save changing again does. +func TestQuietSession_ALeftoverThatTicksOverIsLetGo(t *testing.T) { + d := newTestDaemon(t) + g := quietGame(t, d, "leftover", 30*time.Millisecond) // busy enough to count, every poll + t0 := time.Now() + for i := 0; i <= 5; i++ { + d.pollSessionsAt(t0.Add(time.Duration(i) * 10 * time.Minute)) + } + if d.PlayingSince(g.ID).IsZero() { + t.Fatal("setup: the session ended before the hour") + } + d.pollSessionsAt(t0.Add(61 * time.Minute)) + if !d.PlayingSince(g.ID).IsZero() { + t.Fatal("an hour without a save, all but idle, still holds the save") + } + d.pollSessionsAt(t0.Add(62 * time.Minute)) + if !d.PlayingSince(g.ID).IsZero() { + t.Error("the process still running started the session again at once") + } + if err := d.Store.SetLastManifestHash(g.ID, "a save made since"); err != nil { + t.Fatal(err) + } + d.pollSessionsAt(t0.Add(63 * time.Minute)) + if d.PlayingSince(g.ID).IsZero() { + t.Error("the game saved again but no session started") + } +} + +// A game saved only by hand can go hours without a save while it is played. +// It keeps drawing, so its session goes on. +func TestQuietSession_AGamePlayedWithoutSavingKeepsItsSession(t *testing.T) { + d := newTestDaemon(t) + g := quietGame(t, d, "manual-saves", time.Minute) // a minute of processor time per ten + t0 := time.Now() + for i := 0; i <= 18; i++ { // three hours + d.pollSessionsAt(t0.Add(time.Duration(i) * 10 * time.Minute)) + } + if d.PlayingSince(g.ID).IsZero() { + t.Error("a game being played without saving lost its session") + } +} + +// A session begun from outside — the Deck plugin, from Steam's launch — ends +// when that says so, however quiet the game is. +func TestQuietSession_ASessionBegunFromOutsideIsNotEnded(t *testing.T) { + d := newTestDaemon(t) + g := quietGame(t, d, "deck", 30*time.Millisecond) + t0 := time.Now() + d.sessions.tracker.Begin(g.ID, t0) + for i := 0; i <= 12; i++ { + d.pollSessionsAt(t0.Add(time.Duration(i) * 10 * time.Minute)) + } + if d.PlayingSince(g.ID).IsZero() { + t.Error("a session begun by the Deck plugin was ended as quiet") + } +} diff --git a/internal/p2p/routes.go b/internal/p2p/routes.go index ba2142ff..abbea2b4 100644 --- a/internal/p2p/routes.go +++ b/internal/p2p/routes.go @@ -683,6 +683,11 @@ func (e *Engine) handleManifest(w http.ResponseWriter, r *http.Request) { jsonError(w, http.StatusNotFound, syncengine.HeldMessage) return } + // Being played here: synced when the session ends (syncengine/playing.go). + if e.Sync.PlayingHere(game.ID) { + jsonError(w, http.StatusServiceUnavailable, syncengine.PlayingMessage) + return + } // Never describe a save a sync here is writing: part-way through, it is a // mixture no device holds, and the asker would judge it as a save that had // moved (syncengine/settle.go). Held still while it is read; a write in @@ -811,6 +816,10 @@ func (e *Engine) handleDeleteFile(w http.ResponseWriter, r *http.Request) { jsonError(w, http.StatusNotFound, "Game not found.") return } + if e.Sync.PlayingHere(game.ID) { + jsonError(w, http.StatusServiceUnavailable, syncengine.PlayingMessage) + return + } base, ok := e.resolveServeRoot(gameID, game, body.Root) if !ok { jsonError(w, http.StatusNotFound, "This device has no save location named "+strconv.Quote(body.Root)+" for that game.") diff --git a/internal/p2p/syncengine/engine.go b/internal/p2p/syncengine/engine.go index 3a0089fa..2e376706 100644 --- a/internal/p2p/syncengine/engine.go +++ b/internal/p2p/syncengine/engine.go @@ -160,6 +160,13 @@ type Engine struct { // settings (presets/devicesettings.go), set by the daemon; nil leaves // them syncing. See IgnoreText. DeviceSettingsFor func(store.Game) []string + // Playing reports whether a game is being played on this device + // (playing.go). Set by the daemon from its play sessions. + Playing func(gameID string) bool + // playingMu guards playingSaid: per game and peer, whether its "playing" + // answer was said already this session (playing.go). + playingMu sync.Mutex + playingSaid map[string]bool } // New creates an Engine. @@ -259,6 +266,10 @@ func (e *Engine) SyncGame(ctx context.Context, gameID string, onlinePeers []Peer if e.Paused != nil && e.Paused() { return nil, ErrPaused } + // A game being played here is synced when the session ends (playing.go). + if e.PlayingHere(gameID) { + return nil, ErrPlayingHere + } // A save emptied here is not synced until someone says whether that was // meant (hold.go). held, err := e.checkHoldSettled(ctx, gameID) @@ -362,7 +373,7 @@ func (e *Engine) SyncGame(ctx context.Context, gameID string, onlinePeers []Peer // divergence from the NEXT sync, causing the peer to silently overwrite // its own changes instead of detecting the conflict and asking. switch res.Status { - case "conflict", "error": + case "conflict", "error", "peer_playing": case "peer_missing", "peer_awaiting_folder", "peer_holding": // The two devices talked and finished, which is what the // per-device stamp has always recorded. But nothing of THIS game @@ -456,6 +467,12 @@ func (e *Engine) SyncWithPeer(ctx context.Context, gameID string, peer Peer) (Re if isAwaitingFolder(err) { return Result{Status: "peer_awaiting_folder", PeerID: peer.ID, PeerName: peer.Name}, nil } + // Being played there: its save comes when the session ends, sent by + // that device. Before isSettling, whose message this one contains. + if isPlaying(err) { + e.notePeerPlaying(gameID, peer, game.Name) + return Result{Status: "peer_playing", PeerID: peer.ID, PeerName: peer.Name}, nil + } // The peer is still writing this game for a sync of its own, and // would otherwise have described a save half-way between two states // (settle.go). SyncGame asks again once it has finished. @@ -473,6 +490,7 @@ func (e *Engine) SyncWithPeer(ctx context.Context, gameID string, peer Peer) (Re } return Result{}, fmt.Errorf("fetch remote manifest: %w", err) } + e.peerNotPlaying(gameID, peer.ID) // 2. Branch alignment: local follows the remote's active branch. // diff --git a/internal/p2p/syncengine/playing.go b/internal/p2p/syncengine/playing.go new file mode 100644 index 00000000..5c689617 --- /dev/null +++ b/internal/p2p/syncengine/playing.go @@ -0,0 +1,74 @@ +package syncengine + +import ( + "fmt" + "strings" +) + +// A game being played. +// +// A game autosaves every few minutes, and each autosave is a change of its +// save. Synced as they come, every one of them reached every other device as +// a version of its own, and each of those took a snapshot of it: eleven of +// one game in 26 minutes on a device nobody was sitting at. And this device +// took other devices' versions of a game running here, writing a save under +// a game that was still using it. +// +// So while a game is played here (Playing, from the daemon's sessions), it +// is not synced either way. Another device asking for it is told so, and +// asks again later; nothing is taken for it. When the session ends, the save +// as it was left goes to the other devices in one sync (daemon/sessions.go). + +// PlayingMessage is what a device answers when asked for a game being played +// on it. It contains SettlingMessage, so a device on a build that predates +// this takes it as "busy, ask again" too. +const PlayingMessage = "This device is playing this game right now; its save follows when the session ends. " + SettlingMessage + +const playingMark = "is playing this game right now" + +// isPlaying reports whether an error from a peer is its "playing" answer. +func isPlaying(err error) bool { + return err != nil && strings.Contains(err.Error(), playingMark) +} + +// playingErr is SyncGame declining to sync a game being played here. It +// counts as ErrHeld for callers that only want to know it was not an error. +type playingErr struct{} + +func (playingErr) Error() string { + return "it is being played on this device; it syncs when the session ends" +} + +func (playingErr) Is(target error) bool { return target == ErrHeld || target == ErrPlayingHere } + +// ErrPlayingHere is SyncGame declining to sync a game being played here. +var ErrPlayingHere error = playingErr{} + +// PlayingHere reports whether a game is being played on this device. +func (e *Engine) PlayingHere(gameID string) bool { + return e.Playing != nil && e.Playing(gameID) +} + +// notePeerPlaying says once per session that a peer is playing a game: the +// reconcile asks every minute, and each "playing" answer is the same news. +func (e *Engine) notePeerPlaying(gameID string, peer Peer, gameName string) { + key := gameID + "\x00" + peer.ID + e.playingMu.Lock() + said := e.playingSaid[key] + if e.playingSaid == nil { + e.playingSaid = map[string]bool{} + } + e.playingSaid[key] = true + e.playingMu.Unlock() + if !said { + e.Log("info", fmt.Sprintf("%s is playing %q right now; its save follows when the session ends", peer.Name, gameName)) + } +} + +// peerNotPlaying forgets that a peer was playing a game, once it answers +// normally again: its next session is said again. +func (e *Engine) peerNotPlaying(gameID, peerID string) { + e.playingMu.Lock() + delete(e.playingSaid, gameID+"\x00"+peerID) + e.playingMu.Unlock() +} diff --git a/internal/p2p/syncengine/playing_test.go b/internal/p2p/syncengine/playing_test.go new file mode 100644 index 00000000..90112fb3 --- /dev/null +++ b/internal/p2p/syncengine/playing_test.go @@ -0,0 +1,49 @@ +package syncengine + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" +) + +// A game being played here is not synced either way until the session ends: +// nothing is taken for it, and callers are not told it failed. +func TestPlaying_NotSyncedWhilePlayedHere(t *testing.T) { + env := setupEngine(t) + write(t, env.localDir, "slot1.sav", "being played") + write(t, env.remoteDir, "slot1.sav", "another device's") + playing := true + env.engine.Playing = func(string) bool { return playing } + + _, err := env.engine.SyncGame(context.Background(), "game1", []Peer{env.peer}) + if !errors.Is(err, ErrPlayingHere) || !errors.Is(err, ErrHeld) { + t.Fatalf("err = %v, want ErrPlayingHere, quiet as ErrHeld", err) + } + if got, _ := os.ReadFile(filepath.Join(env.localDir, "slot1.sav")); string(got) != "being played" { + t.Fatalf("the save of a running game was replaced: %q", got) + } + + playing = false + if _, err := env.engine.SyncGame(context.Background(), "game1", []Peer{env.peer}); err != nil { + t.Fatalf("after the session: %v", err) + } +} + +// Another device playing the game answers "playing": nothing is compared or +// stamped, and asking does not count as a failure. +func TestPlaying_PeerPlayingIsWaitedFor(t *testing.T) { + env := setupEngine(t) + write(t, env.localDir, "slot1.sav", "here") + env.transport.manifestErr = errors.New(PlayingMessage) + + res, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer) + if err != nil || res.Status != "peer_playing" { + t.Fatalf("status = %q, %v; want peer_playing", res.Status, err) + } + // A device on an older build reads the same answer as "busy". + if !isSettling(errors.New(PlayingMessage)) { + t.Error("the playing answer is not recognisable as busy by older builds") + } +} diff --git a/internal/p2p/wanclient_handlers.go b/internal/p2p/wanclient_handlers.go index 9a899c0c..24acf679 100644 --- a/internal/p2p/wanclient_handlers.go +++ b/internal/p2p/wanclient_handlers.go @@ -616,6 +616,9 @@ func (w *WanClient) serveManifest(route string, body json.RawMessage, peerID str if w.engine.holdingBack(game) { return 404, map[string]string{"error": syncengine.HeldMessage} } + if w.engine.Sync.PlayingHere(game.ID) { + return 503, map[string]string{"error": syncengine.PlayingMessage} + } // As on the LAN: a save a sync here is writing is not described // (syncengine/settle.go). Each request runs on its own goroutine, so the // wait does not hold up the relay traffic that finishes the write. @@ -702,6 +705,9 @@ func (w *WanClient) serveDeleteFile(route string, rawBody json.RawMessage, fromP if err != nil { return 404, map[string]string{"error": "Game not found."} } + if w.engine.Sync.PlayingHere(game.ID) { + return 503, map[string]string{"error": syncengine.PlayingMessage} + } if !delta.IsSafePath(game.SavePath, body.RelPath) { return 403, map[string]string{"error": "invalid path"} } diff --git a/internal/sessions/procs_windows.go b/internal/sessions/procs_windows.go index 1fb4ceb8..6e33c11c 100644 --- a/internal/sessions/procs_windows.go +++ b/internal/sessions/procs_windows.go @@ -3,13 +3,14 @@ package sessions import ( + "time" "unsafe" "golang.org/x/sys/windows" ) // listProcesses walks the process table and asks each process for its full -// program path. A process this user may not look at (a service, another +// program path and how much processor time it has used. A process this user may not look at (a service, another // user's) is skipped: no game of theirs is one of ours. func listProcesses() ([]Proc, error) { snap, err := windows.CreateToolhelp32Snapshot(windows.TH32CS_SNAPPROCESS, 0) @@ -27,8 +28,8 @@ func listProcesses() ([]Proc, error) { buf := make([]uint16, windows.MAX_LONG_PATH) for { if pid := entry.ProcessID; pid > 4 { - if exe := imagePath(pid, buf); exe != "" { - out = append(out, Proc{PID: int(pid), Exe: exe}) + if p, ok := inspect(pid, buf); ok { + out = append(out, p) } } if err := windows.Process32Next(snap, &entry); err != nil { @@ -38,15 +39,26 @@ func listProcesses() ([]Proc, error) { return out, nil } -func imagePath(pid uint32, buf []uint16) string { +func inspect(pid uint32, buf []uint16) (Proc, bool) { h, err := windows.OpenProcess(windows.PROCESS_QUERY_LIMITED_INFORMATION, false, pid) if err != nil { - return "" + return Proc{}, false } defer windows.CloseHandle(h) n := uint32(len(buf)) if err := windows.QueryFullProcessImageName(h, 0, &buf[0], &n); err != nil { - return "" + return Proc{}, false } - return windows.UTF16ToString(buf[:n]) + p := Proc{PID: int(pid), Exe: windows.UTF16ToString(buf[:n])} + + var creation, exit, kernel, user windows.Filetime + timesOK := windows.GetProcessTimes(h, &creation, &exit, &kernel, &user) == nil + if timesOK { + p.Measured = true + // FILETIME counts 100-nanosecond intervals. + ticks := (uint64(kernel.HighDateTime)<<32 | uint64(kernel.LowDateTime)) + + (uint64(user.HighDateTime)<<32 | uint64(user.LowDateTime)) + p.CPU = time.Duration(ticks * 100) + } + return p, true } diff --git a/internal/sessions/sessions.go b/internal/sessions/sessions.go index 1f47e5c3..9b586c1b 100644 --- a/internal/sessions/sessions.go +++ b/internal/sessions/sessions.go @@ -34,6 +34,12 @@ type Proc struct { Exe string // Args is the command line, where the system says (Linux). Args []string + // Measured says CPU was read (Windows): the processor time the program has + // used so far. A game being played uses some all the time; what a game + // leaves behind when it closes — a crash handler, a stub of the same name — + // uses none (daemon.activeProcesses). + Measured bool + CPU time.Duration // SteamAppID is the Steam app the process was started for, where Steam // tagged it (Linux: SteamAppId in the environment, or AppId=… on the // command line of Steam's launcher). @@ -252,3 +258,40 @@ func (t *Tracker) Finish(gameID string, at time.Time) { // List returns the processes running now. Where the system cannot say, it // returns none and an error, and nothing is ever seen playing. func List() ([]Proc, error) { return listProcesses() } + +// GameCPU is the processor time a game's running processes have used so far, +// summed, and whether every one of them could be measured. +type GameCPU struct { + CPU time.Duration + Measured bool +} + +// CPUOf sums, for each target with a process running, the processor time its +// processes have used, matched as Running matches them. +func CPUOf(procs []Proc, targets []Target) map[string]GameCPU { + out := map[string]GameCPU{} + for _, t := range targets { + for _, p := range procs { + if !matches(p, t) { + continue + } + g, seen := out[t.GameID] + if !seen { + g.Measured = true + } + g.CPU += p.CPU + g.Measured = g.Measured && p.Measured + out[t.GameID] = g + } + } + return out +} + +// Marked reports whether a game's session was begun from outside (Begin) — +// the Deck plugin, `opensave wrap` — and so ends only when told. +func (t *Tracker) Marked(gameID string) bool { + t.mu.Lock() + defer t.mu.Unlock() + s, ok := t.playing[gameID] + return ok && s.marked +} diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go index a5477f3f..e4ad729f 100644 --- a/internal/watcher/watcher.go +++ b/internal/watcher/watcher.go @@ -131,6 +131,11 @@ type Callbacks struct { // 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 + // HoldSnapshot reports whether a change should not be snapshotted now: + // the game is being played, and a checkpoint was taken recently. The + // change is not recorded as captured either, so the next one, or the end + // of the session, still sees it. May be nil. + HoldSnapshot func(gameID string) bool } // syncWriting reports whether a sync is or was just writing a game @@ -979,6 +984,9 @@ func (e *Engine) handleChangeFrom(ctx context.Context, gw *gameWatch, ownOnly bo } return } + if e.cb.HoldSnapshot != nil && e.cb.HoldSnapshot(gw.gameID) { + return // being played: kept by the session (a checkpoint, or its end) + } for attempt := 1; attempt <= snapshotMaxRetries; attempt++ { err = e.cb.CreateSnapshot(gw.gameID)