From 704c9d82981fa020422968ca8991231714d75d7c Mon Sep 17 00:00:00 2001 From: ChakraFusion <112588066+ChakraFusion@users.noreply.github.com> Date: Thu, 1 Oct 2026 12:00:24 +0200 Subject: [PATCH 1/4] sync: batch small files, skip unchanged manifests, probe presence on its own Performance - Batched pulls (protocol revision 2): small files are fetched up to 256 per request (4MB), 4 requests in flight, over a new /api/p2p/files route. Each file is still written through the PatchWriter (verified, fsynced, renamed). One request per file cost ~200ms on a real LAN; on loopback 5000 files went from 42s to 12s. Peers below revision 2 and relay connections keep the per-file path. - Unchanged games: when this device still holds the agreed base, the manifest request carries ifHash and a peer that still holds it too answers "unchanged" instead of sending the manifest. Older peers ignore the parameter; their full answer is used, not fetched twice. - Manifests are gzipped for clients that accept it (every Go client does, so older versions read them too): ~5.6x smaller with real hashes. - Manifests and batches get a 10-minute client timeout instead of 30s; a large manifest over a VPN never arrived inside 30s, so those games never synced. - An in-sync pass also mirrors the peer's latest snapshot, which only a pull did before; games identical everywhere from the start showed history on one device only. - A batch is bounded by what the responder counts (each requested block at the file's full block size, at most the file's real size), so a batch of many small files with a few larger ones is never refused as too large; a batch refused anyway is fetched again as two halves. Robustness - Presence: paired LAN peers are probed every 10s on their own ticker. Probes used to ride along with the minute-long reconcile, which waits for a whole sync-all pass; peers reached over a VPN (no UDP broadcast) showed online/offline minutes late. Probe rounds are serialized. - Games whose save folder is not on this device are skipped by sync-all and dropped from the retry queue instead of failing every 20s. - "syncing X with Y" is logged only when there is something to do; most passes produced only "already in sync" lines, which rotated everything else out of the log within hours. - The interrupted-pull safety test also runs over batched pulls. Co-Authored-By: Claude Opus 5.5 --- e2e/batch_bench_test.go | 59 ++++ internal/p2p/engine.go | 76 +++++ internal/p2p/manifest_gzip_test.go | 63 ++++ internal/p2p/missing_skip_test.go | 28 ++ internal/p2p/presence_test.go | 82 +++++ internal/p2p/routes.go | 139 +++++++- internal/p2p/syncengine/batch.go | 296 ++++++++++++++++++ internal/p2p/syncengine/batch_test.go | 237 ++++++++++++++ internal/p2p/syncengine/engine.go | 63 +++- internal/p2p/syncengine/engine_test.go | 53 ++++ .../p2p/syncengine/partial_safety_test.go | 14 +- internal/p2p/syncengine/quick.go | 73 +++++ internal/p2p/syncengine/quick_test.go | 117 +++++++ internal/p2p/syncengine/transport.go | 43 +++ internal/p2p/transport_lan.go | 65 +++- internal/p2p/transport_wan.go | 14 + 16 files changed, 1410 insertions(+), 12 deletions(-) create mode 100644 e2e/batch_bench_test.go create mode 100644 internal/p2p/manifest_gzip_test.go create mode 100644 internal/p2p/missing_skip_test.go create mode 100644 internal/p2p/presence_test.go create mode 100644 internal/p2p/syncengine/batch.go create mode 100644 internal/p2p/syncengine/batch_test.go create mode 100644 internal/p2p/syncengine/quick.go create mode 100644 internal/p2p/syncengine/quick_test.go diff --git a/e2e/batch_bench_test.go b/e2e/batch_bench_test.go new file mode 100644 index 00000000..0299c4e8 --- /dev/null +++ b/e2e/batch_bench_test.go @@ -0,0 +1,59 @@ +package e2e + +import ( + "fmt" + "net/http" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "github.com/opensave/opensave/internal/p2p" + "github.com/opensave/opensave/internal/p2p/syncengine" + "github.com/opensave/opensave/testutil" +) + +// TestBench_ManySmallFiles times pulling a save of many small files between +// two real daemons, batched and file by file. Opt-in, because it is a +// measurement rather than a check: OPENSAVE_BENCH_FILES=5000 go test ./e2e -run TestBench -v +func TestBench_ManySmallFiles(t *testing.T) { + n, _ := strconv.Atoi(os.Getenv("OPENSAVE_BENCH_FILES")) + if n <= 0 { + t.Skip("set OPENSAVE_BENCH_FILES to run") + } + for _, mode := range []struct { + name string + proto int + }{{"batched", syncengine.ProtoBatchFiles}, {"file-by-file", syncengine.ProtoMultiRoot}} { + t.Run(mode.name, func(t *testing.T) { + restore := p2p.SetServedProto(mode.proto) + defer p2p.SetServedProto(restore) + + a := testutil.NewTestDaemon(t, "Bench-A") + b := testutil.NewTestDaemon(t, "Bench-B") + a.PairWith(b) + chunk := strings.Repeat("x", 3500) // a Project Zomboid map chunk is ~3.5KB + for i := 0; i < n; i++ { + a.WriteSave(fmt.Sprintf("map/map_%d_%d.bin", i/100, i%100), chunk+strconv.Itoa(i)) + } + restoreSyncOnTrack := suppressSyncOnTrack(a, b) + gameID := a.TrackGame("Bench") + b.API(http.MethodPost, "/api/games", map[string]string{"name": "Bench", "savePath": b.SaveDir}, nil) + restoreSyncOnTrack() + + start := time.Now() + b.API(http.MethodPost, "/api/games/"+gameID+"/sync", nil, nil) + count := func() int { + m, _ := filepath.Glob(filepath.Join(b.SaveDir, "map", "*.bin")) + return len(m) + } + if !testutil.WaitFor(60*time.Minute, func() bool { return count() >= n }) { + t.Fatalf("only %d of %d files arrived", count(), n) + } + el := time.Since(start) + t.Logf("%s: %d files in %s = %.2f ms/file", mode.name, n, el.Round(time.Millisecond), float64(el.Microseconds())/1000/float64(n)) + }) + } +} diff --git a/internal/p2p/engine.go b/internal/p2p/engine.go index b0029623..203407df 100644 --- a/internal/p2p/engine.go +++ b/internal/p2p/engine.go @@ -9,11 +9,15 @@ import ( "encoding/json" "errors" "fmt" + "io/fs" "net/http" + "os" + "path/filepath" "strings" "sync" "time" + "github.com/opensave/opensave/internal/delta" "github.com/opensave/opensave/internal/e2ee" "github.com/opensave/opensave/internal/p2p/discovery" "github.com/opensave/opensave/internal/p2p/pairing" @@ -119,6 +123,8 @@ type Engine struct { // that a device has gone away — see offlineStrikes. pingMu sync.Mutex pingMisses map[string]int + // probeMu serializes PingPairedPeers rounds. + probeMu sync.Mutex // Lineage refreshes already running after a peer-applied deletion, keyed // by game+peer. Deletions arrive one file at a time, and each one leaves @@ -386,6 +392,11 @@ func (e *Engine) clearPingMisses(peerID string) { // PingPairedPeers probes every paired LAN peer and updates their // online/offline status. func (e *Engine) PingPairedPeers(ctx context.Context) { + // One probe round at a time: the presence loop and a sync starting can + // both ask, and two overlapping rounds would count one outage as two + // misses toward offlineStrikes. + e.probeMu.Lock() + defer e.probeMu.Unlock() peers, err := e.Store.ListPeers() if err != nil { return @@ -546,6 +557,8 @@ func (e *Engine) StartResyncLoop() { stop := e.stopRetry e.pendingMu.Unlock() + e.startPresenceLoop(stop) + // Tracked on the engine's lifecycle so a shutdown lands between ticks // rather than half-way through a transfer. e.GoSync(func(ctx context.Context) { @@ -575,6 +588,40 @@ func (e *Engine) StartResyncLoop() { }) } +// presenceInterval is how often paired LAN peers are probed just to know +// whether they are there. +// +// Probing used to happen only as a side effect: before the 60-second +// reconcile, before a retry, before a manual sync. The reconcile runs on the +// same goroutine as the sync-everything pass it starts, so while a big +// library synced the next probe waited minutes, and with offlineStrikes +// that was minutes more before a device that had gone showed as gone. Peers +// seen by UDP broadcast hid this; one reached over a VPN such as Tailscale, +// which does not carry broadcasts, depends on probes alone. +// +// 10s makes "back" show within ~10s and "gone" within ~30s. A probe is one +// small GET per peer. A var only so a test can shorten it. +var presenceInterval = 10 * time.Second + +// startPresenceLoop probes paired peers on its own ticker, independent of +// whatever sync work is running. It stops with the resync loop. +func (e *Engine) startPresenceLoop(stop chan struct{}) { + e.GoSync(func(ctx context.Context) { + ticker := time.NewTicker(presenceInterval) + defer ticker.Stop() + for { + select { + case <-stop: + return + case <-ctx.Done(): + return + case <-ticker.C: + e.PingPairedPeers(ctx) + } + } + }) +} + // reconcileAllGames refreshes peer status and re-syncs every game — the // periodic backstop against missed sync triggers. func (e *Engine) reconcileAllGames(ctx context.Context) { @@ -606,6 +653,13 @@ func (e *Engine) retryPendingResyncs(ctx context.Context) { if ctx.Err() != nil { return // shutting down; the failsafe picks these up next start } + // A save folder that is not here cannot be synced, and retrying it + // every tick only logs the same failure. Dropped from the queue; the + // periodic reconcile picks it up once the folder is back. + if g, err := e.Store.GetGame(id); err == nil && saveFolderAbsent(g.SavePath) { + e.ClearPendingResync(id) + continue + } e.Log("info", fmt.Sprintf("retrying interrupted sync for %s", id)) results, err := e.Sync.SyncGame(ctx, id, online) if err == nil { @@ -620,6 +674,22 @@ func (e *Engine) retryPendingResyncs(ctx context.Context) { } } +// saveFolderAbsent reports whether a game's save folder is not on this device +// — the same test as daemon.SaveFolderMissing, which this package cannot +// import. For a single-file save it is the folder the file goes in that +// counts: the file itself may not have been written yet. +func saveFolderAbsent(savePath string) bool { + if savePath == "" { + return false + } + folder := savePath + if isFile, err := delta.ResolveLocalSaveFilePath(savePath); err == nil && isFile { + folder = filepath.Dir(savePath) + } + _, err := os.Stat(folder) + return errors.Is(err, fs.ErrNotExist) +} + // SyncAllGames syncs every tracked game (used when a peer comes online). func (e *Engine) SyncAllGames(ctx context.Context) { if e.Pause.Paused() { @@ -637,6 +707,12 @@ func (e *Engine) SyncAllGames(ctx context.Context) { if !g.AutoSync { continue } + // Not on this device (a drive it doesn't have, a folder that is + // gone): nothing to sync, and it is shown as such on screen. Trying + // anyway logged an error for it against every peer, every pass. + if saveFolderAbsent(g.SavePath) { + continue + } results, err := e.Sync.SyncGame(ctx, g.ID, online) if errors.Is(err, syncengine.ErrHeld) { continue // said once, when it was held; asked about on screen diff --git a/internal/p2p/manifest_gzip_test.go b/internal/p2p/manifest_gzip_test.go new file mode 100644 index 00000000..cca409cf --- /dev/null +++ b/internal/p2p/manifest_gzip_test.go @@ -0,0 +1,63 @@ +package p2p + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + + "github.com/opensave/opensave/internal/delta" + "github.com/opensave/opensave/internal/p2p/syncengine" +) + +// A large manifest goes over the wire gzipped, and an ordinary Go client — +// which is what every OpenSave version uses — reads it without knowing. +func TestManifestIsCompressedAndReadable(t *testing.T) { + m := delta.Manifest{Files: map[string]delta.FileEntry{}} + for i := 0; i < 5000; i++ { + sum := sha256.Sum256([]byte(fmt.Sprint(i))) // random-looking, like real hashes + h := hex.EncodeToString(sum[:]) + m.Files[fmt.Sprintf("Saves/world/map_%d_%d.bin", i/100, i%100)] = delta.FileEntry{ + Size: 3500, Hash: h, BlockSize: 65536, Blocks: []delta.Block{{Index: 0, Hash: h, Length: 3500}}, + } + } + resp := syncengine.ManifestResponse{Manifest: m, ActiveBranch: "main"} + + var wire atomic.Int64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cw := &countingResponse{ResponseWriter: w} + jsonOKCompressed(cw, r, resp) + wire.Store(cw.n) + })) + defer srv.Close() + + req, _ := http.NewRequest(http.MethodGet, srv.URL, nil) + var got syncengine.ManifestResponse + if err := doJSONWith(lanBulkClient, req, &got); err != nil { + t.Fatal(err) + } + if len(got.Manifest.Files) != len(m.Files) || got.Manifest.ManifestHash() != m.ManifestHash() { + t.Fatalf("manifest came back with %d files and a different hash", len(got.Manifest.Files)) + } + + plain := httptest.NewRecorder() + jsonOK(plain, resp) + if wire.Load()*2 > int64(plain.Body.Len()) { + t.Errorf("compressed %d bytes vs %d plain — expected at least 2x smaller", wire.Load(), plain.Body.Len()) + } + t.Logf("manifest of %d files: %d bytes plain, %d on the wire", len(m.Files), plain.Body.Len(), wire.Load()) +} + +type countingResponse struct { + http.ResponseWriter + n int64 +} + +func (c *countingResponse) Write(b []byte) (int, error) { + n, err := c.ResponseWriter.Write(b) + c.n += int64(n) + return n, err +} diff --git a/internal/p2p/missing_skip_test.go b/internal/p2p/missing_skip_test.go new file mode 100644 index 00000000..b5de9828 --- /dev/null +++ b/internal/p2p/missing_skip_test.go @@ -0,0 +1,28 @@ +package p2p + +import ( + "os" + "path/filepath" + "testing" +) + +func TestSaveFolderAbsent(t *testing.T) { + dir := t.TempDir() + if saveFolderAbsent(dir) { + t.Error("an existing folder was reported absent") + } + if !saveFolderAbsent(filepath.Join(dir, "gone")) { + t.Error("a missing folder was not reported absent") + } + // A single-file save not written yet: its folder is what counts. + file := filepath.Join(dir, "save.sav") + if err := os.WriteFile(file, []byte("x"), 0o666); err != nil { + t.Fatal(err) + } + if saveFolderAbsent(file) { + t.Error("an existing single-file save was reported absent") + } + if saveFolderAbsent("") { + t.Error("an empty path was reported absent") + } +} diff --git a/internal/p2p/presence_test.go b/internal/p2p/presence_test.go new file mode 100644 index 00000000..de0397de --- /dev/null +++ b/internal/p2p/presence_test.go @@ -0,0 +1,82 @@ +package p2p + +import ( + "net" + "net/http" + "net/http/httptest" + "path/filepath" + "strconv" + "sync/atomic" + "testing" + "time" + + "github.com/opensave/opensave/internal/snapshot" + "github.com/opensave/opensave/internal/store" +) + +func waitUntil(d time.Duration, cond func() bool) bool { + deadline := time.Now().Add(d) + for time.Now().Before(deadline) { + if cond() { + return true + } + time.Sleep(20 * time.Millisecond) + } + return cond() +} + +// A peer's online state follows it on its own ticker, whatever else the +// engine is doing — not only when a sync or the minute-long reconcile happens +// to probe. A peer reached over a VPN gets no UDP broadcasts, so without this +// its indicator lagged by minutes. +func TestPresenceLoop_FollowsPeer(t *testing.T) { + old := presenceInterval + presenceInterval = 50 * time.Millisecond + t.Cleanup(func() { presenceInterval = old }) + + var up atomic.Bool + up.Store(true) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !up.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"status":"ok","appVersion":"2.4.0"}`)) + })) + t.Cleanup(srv.Close) + host, portStr, _ := net.SplitHostPort(srv.Listener.Addr().String()) + port, _ := strconv.Atoi(portStr) + + s, err := store.Open(filepath.Join(t.TempDir(), "opensave.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { s.Close() }) + if err := s.EnsureDefaultSettings(t.TempDir(), t.TempDir()); err != nil { + t.Fatal(err) + } + if err := s.UpsertPeer(store.Peer{ID: "node_vpn", Name: "Over VPN", Address: host, Port: port, Status: "offline"}); err != nil { + t.Fatal(err) + } + + e := New(s, snapshot.New(s), func(string, string) {}) + stop := make(chan struct{}) + e.startPresenceLoop(stop) + t.Cleanup(func() { close(stop); e.cancel(); e.bgSyncWG.Wait() }) + + status := func() string { + p, err := s.GetPeer("node_vpn") + if err != nil { + return "" + } + return p.Status + } + if !waitUntil(3*time.Second, func() bool { return status() == "online" }) { + t.Fatalf("peer answering pings still %q", status()) + } + up.Store(false) + if !waitUntil(3*time.Second, func() bool { return status() == "offline" }) { + t.Fatalf("peer no longer answering still %q", status()) + } +} diff --git a/internal/p2p/routes.go b/internal/p2p/routes.go index 9d8d1c73..666be2c6 100644 --- a/internal/p2p/routes.go +++ b/internal/p2p/routes.go @@ -1,6 +1,7 @@ package p2p import ( + "compress/gzip" "context" "encoding/json" "errors" @@ -11,6 +12,7 @@ import ( "os" "path/filepath" "strconv" + "strings" "sync/atomic" "time" @@ -47,6 +49,7 @@ func (e *Engine) RegisterRoutes(r chi.Router) { r.Use(e.refuseWhilePaused) r.Get("/api/p2p/manifest/{gameId}", e.handleManifest) r.Post("/api/p2p/blocks/{gameId}", e.handleBlocks) + r.Post("/api/p2p/files/{gameId}", e.handleFileBatch) r.Post("/api/p2p/delete-file/{gameId}", e.handleDeleteFile) r.Get("/api/sync/trigger/{gameId}", e.handleSyncTrigger) }) @@ -725,7 +728,19 @@ func (e *Engine) handleManifest(w http.ResponseWriter, r *http.Request) { if latest, err := e.Snapshots.LatestSnapshot(gameID, ""); err == nil { resp.LatestSnapshot = &syncengine.SnapshotInfo{ID: latest.ID, Timestamp: latest.Timestamp, Comment: latest.Comment} } - jsonOK(w, resp) + // The asker already holds the state both devices last agreed on and says + // so; if this device still holds it too, "unchanged" is the whole answer. + // For a save of a quarter-million files the full manifest is tens of MB + // of JSON, and the minute-long reconcile used to fetch it from every peer + // every minute to learn that nothing had changed. Single-location games + // only: extra locations carry their own bases (see RootHash). + if ifHash := r.URL.Query().Get("ifHash"); ifHash != "" && len(manifest.Extra) == 0 && + manifest.ManifestHash() == ifHash { + resp.Manifest = delta.Manifest{} + resp.Unchanged = true + resp.ManifestHash = ifHash + } + jsonOKCompressed(w, r, resp) } // holdForServing holds the game's save still for a manifest to be served from @@ -788,6 +803,100 @@ func (e *Engine) handleBlocks(w http.ResponseWriter, r *http.Request) { jsonOK(w, map[string]any{"blocks": encodeBlocks(blocks, wantsGzip(body.Encodings))}) } +// Bounds on one batch request, so a peer cannot make this device read and +// hold an unbounded amount in a single response. The requester stays well +// under them (syncengine.batchMaxFiles / batchMaxBytes). +const ( + fileBatchMaxFiles = 1024 + fileBatchMaxBytes = 16 << 20 +) + +// handleFileBatch serves the blocks of many files of one save location in a +// single response: handleBlocks for a list. One round trip per file was what +// made a save of many small files take hours; see syncengine.ProtoBatchFiles. +// +// A file that cannot be read is reported in its own entry rather than failing +// the batch, so the requester can say which one it was. +func (e *Engine) handleFileBatch(w http.ResponseWriter, r *http.Request) { + gameID := chi.URLParam(r, "gameId") + var body struct { + Root string `json:"root"` + Files []syncengine.FileBlocksRequest `json:"files"` + Encodings []string `json:"encodings"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil || len(body.Files) == 0 { + jsonError(w, http.StatusBadRequest, "files are required") + return + } + if len(body.Files) > fileBatchMaxFiles { + jsonError(w, http.StatusBadRequest, fmt.Sprintf("at most %d files per batch", fileBatchMaxFiles)) + return + } + game, err := e.trackedGameForPeer(gameID) + if err != nil { + jsonError(w, http.StatusNotFound, "Game not found.") + 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.") + return + } + singleFile, _ := delta.ResolveLocalSaveFilePath(base) + + // What the answer will hold: each file's requested blocks at its block + // size, but never more than the file itself. Counting every block at full + // size took 256 files of 4 KB for 16 MB — the whole limit — and refused + // any batch of small files with a few larger ones among them, which failed + // whole pulls of saves made of many small files. + var requested int64 + for _, f := range body.Files { + bs := f.BlockSize + if bs <= 0 { + bs = 64 * 1024 + } + n := int64(len(f.BlockIndices)) * int64(bs) + if f.RelPath != "" && delta.IsSafePath(base, f.RelPath) { + p := delta.LocalNameFor(base, f.RelPath) + if singleFile { + p = base + } + if info, err := os.Stat(p); err == nil && info.Size() < n { + n = info.Size() + } + } + requested += n + } + if requested > fileBatchMaxBytes { + jsonError(w, http.StatusBadRequest, "batch too large") + return + } + gz := wantsGzip(body.Encodings) + + out := make([]syncengine.FileBlocks, 0, len(body.Files)) + for _, f := range body.Files { + entry := syncengine.FileBlocks{RelPath: f.RelPath} + if f.RelPath == "" || !delta.IsSafePath(base, f.RelPath) { + entry.Error = "invalid path" + out = append(out, entry) + continue + } + // Resolved, not joined — see handleBlocks. + fullPath := delta.LocalNameFor(base, f.RelPath) + if singleFile { + fullPath = base + } + blocks, err := delta.ReadBlocks(fullPath, f.BlockIndices, f.BlockSize) + if err != nil { + entry.Error = "read blocks failed: " + err.Error() + } else { + entry.Blocks = encodeBlocks(blocks, gz) + } + out = append(out, entry) + } + jsonOK(w, map[string]any{"files": out}) +} + func (e *Engine) handleDeleteFile(w http.ResponseWriter, r *http.Request) { gameID := chi.URLParam(r, "gameId") var body struct { @@ -996,6 +1105,32 @@ func progressEventFromMap(data map[string]any) syncengine.ProgressEvent { return ev } +// jsonOKCompressed is jsonOK, gzipped when the asker accepts it. +// +// For manifests. A save of a few hundred thousand files makes a manifest of +// tens of MB of JSON, which over a VPN did not arrive inside a peer's request +// timeout — so the peer never learned what this device holds, every sync of +// that game failed, and the two devices never converged. Manifest JSON (paths +// and hex hashes) compresses several times over. Go's HTTP client asks for +// gzip and unpacks it on its own unless told not to, so every version of +// OpenSave reads these, including ones that predate this. +func jsonOKCompressed(w http.ResponseWriter, r *http.Request, v any) { + if !strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") { + jsonOK(w, v) + return + } + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Content-Encoding", "gzip") + w.Header().Add("Vary", "Accept-Encoding") + w.WriteHeader(http.StatusOK) + zw, err := gzip.NewWriterLevel(w, gzip.BestSpeed) + if err != nil { + return + } + _ = json.NewEncoder(zw).Encode(v) + _ = zw.Close() +} + func jsonOK(w http.ResponseWriter, v any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) @@ -1036,7 +1171,7 @@ func (e *Engine) resolveServeRoot(gameID string, game store.Game, root string) ( // argument — the read still happens concurrently. var servedProto atomic.Int64 -func init() { servedProto.Store(syncengine.ProtoMultiRoot) } +func init() { servedProto.Store(syncengine.ProtoBatchFiles) } // ServedProto reports the protocol revision advertised to peers. func ServedProto() int { return int(servedProto.Load()) } diff --git a/internal/p2p/syncengine/batch.go b/internal/p2p/syncengine/batch.go new file mode 100644 index 00000000..7a15d94e --- /dev/null +++ b/internal/p2p/syncengine/batch.go @@ -0,0 +1,296 @@ +package syncengine + +import ( + "context" + "errors" + "fmt" + "os" + "strings" + "sync" + "time" + + "github.com/opensave/opensave/internal/delta" +) + +// Pulling many small files, many to a request. +// +// A pull used to cost one HTTP round trip per file, one after the other. +// Measured on a Windows LAN, that came to ~200ms a file, of which writing the +// file (temp, fsync, rename) was ~2ms — so a save made of a quarter-million +// small files took half a day to move 50MB. Batching removes the round trips; +// the writing is the same as before, file by file, and just as safe: each +// file still goes through a PatchWriter (temp file, content verified against +// the manifest's hash, fsync, rename) and gets the peer's mtime. + +const ( + // batchFileMaxBytes is the most a file may need fetched to go into a + // batch. Anything bigger keeps the per-file path, which already fetches + // its blocks in parallel and has nothing to gain here. + batchFileMaxBytes = 256 << 10 + // batchMaxFiles and batchMaxBytes bound one request, well inside what the + // responder accepts (p2p.fileBatchMaxFiles / fileBatchMaxBytes). + batchMaxFiles = 256 + batchMaxBytes = 4 << 20 + // batchMaxClaimed bounds a request as the responder measures it: every + // block requested at the file's full block size, whatever the file's real + // size. 256 files of 4 KB claim 16 MB that way, which is the responder's + // whole limit (p2p.fileBatchMaxBytes), so any file of two blocks in such + // a batch had it refused as "batch too large" — and the whole pull with + // it, over and over, on saves of many small files. Kept under the limit + // with room to spare. + batchMaxClaimed = 12 << 20 +) + +// claimedBytes is what a file costs a batch as the responder counts it. +func claimedBytes(f delta.FileEntry, indices []int) int64 { + bs := f.BlockSize + if bs <= 0 { + bs = 64 * 1024 + } + return int64(len(indices)) * int64(bs) +} + +// batchWorkers is how many batch requests are in flight at once: enough to +// keep the link and the responder's disk busy while this side writes. A var +// only so a test can make the order deterministic. +var batchWorkers = 4 + +type batchJob struct { + relPath string + localPath string + remote delta.FileEntry + indices []int +} + +// changedBytes is how much of a file a pull has to fetch. +func changedBytes(f delta.FileEntry, indices []int) int64 { + var n int64 + for _, idx := range indices { + if idx >= 0 && idx < len(f.Blocks) { + n += int64(f.Blocks[idx].Length) + } + } + return n +} + +// groupBatches splits jobs into requests of at most batchMaxFiles files, +// batchMaxBytes of blocks, and batchMaxClaimed as the responder counts them. +func groupBatches(jobs []batchJob) [][]batchJob { + var out [][]batchJob + var cur []batchJob + var curBytes, curClaimed int64 + for _, j := range jobs { + b := changedBytes(j.remote, j.indices) + c := claimedBytes(j.remote, j.indices) + if len(cur) > 0 && (len(cur) >= batchMaxFiles || curBytes+b > batchMaxBytes || curClaimed+c > batchMaxClaimed) { + out = append(out, cur) + cur, curBytes, curClaimed = nil, 0, 0 + } + cur = append(cur, j) + curBytes += b + curClaimed += c + } + if len(cur) > 0 { + out = append(out, cur) + } + return out +} + +// pullBatched fetches and writes jobs, many files per request. It returns the +// paths it wrote, including on error, since those files are on disk. +func (e *Engine) pullBatched(ctx context.Context, batcher BatchFetcher, peer Peer, gameID, root string, + jobs []batchJob, throttle *throttler, tracker *progressTracker, onProgress func(force bool)) ([]string, error) { + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + var ( + mu sync.Mutex + written []string + firstErr error + ) + fail := func(err error) { + mu.Lock() + if firstErr == nil { + firstErr = err + cancel() + } + mu.Unlock() + } + + batches := groupBatches(jobs) + workers := batchWorkers + if workers > len(batches) { + workers = len(batches) + } + work := make(chan []batchJob) + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for batch := range work { + done, err := e.pullOneBatch(ctx, batcher, peer, gameID, root, batch, throttle, tracker, onProgress) + mu.Lock() + written = append(written, done...) + mu.Unlock() + if err != nil { + fail(err) + return + } + // One line per batch instead of one per file: a quarter-million + // "file updated" lines rotated every other message out of the log. + if len(done) > 0 { + e.Log("info", fmt.Sprintf("%d files updated (%s … %s)", len(done), done[0], done[len(done)-1])) + } + onProgress(true) + } + }() + } + for _, b := range batches { + select { + case work <- b: + case <-ctx.Done(): + } + } + close(work) + wg.Wait() + + mu.Lock() + defer mu.Unlock() + if firstErr != nil { + return written, firstErr + } + return written, ctx.Err() +} + +// pullOneBatch fetches one batch and writes its files. +func (e *Engine) pullOneBatch(ctx context.Context, batcher BatchFetcher, peer Peer, gameID, root string, + batch []batchJob, throttle *throttler, tracker *progressTracker, onProgress func(force bool)) ([]string, error) { + + req := make([]FileBlocksRequest, len(batch)) + for i, j := range batch { + req[i] = FileBlocksRequest{RelPath: j.relPath, BlockIndices: j.indices, BlockSize: j.remote.BlockSize} + } + + var resp []FileBlocks + var err error + const maxAttempts = 3 + for attempt := 1; attempt <= maxAttempts; attempt++ { + resp, err = batcher.FetchFileBatch(ctx, peer, gameID, root, req) + if err == nil || errors.Is(err, ErrBatchUnsupported) { + break + } + // Refused as too large — a responder that counts or limits batches + // differently: the same request will be refused again, so it goes + // as two halves instead. Asking again unchanged failed the whole pull. + if strings.Contains(err.Error(), "batch too large") && len(batch) > 1 { + half := len(batch) / 2 + first, err := e.pullOneBatch(ctx, batcher, peer, gameID, root, batch[:half], throttle, tracker, onProgress) + if err != nil { + return first, err + } + second, err := e.pullOneBatch(ctx, batcher, peer, gameID, root, batch[half:], throttle, tracker, onProgress) + return append(first, second...), err + } + e.Log("warn", fmt.Sprintf("batch fetch attempt %d/%d of %d files failed: %v", attempt, maxAttempts, len(batch), err)) + if attempt < maxAttempts { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(time.Duration(attempt) * time.Second): + } + } + } + if errors.Is(err, ErrBatchUnsupported) { + return e.pullBatchOneByOne(ctx, peer, gameID, root, batch, throttle, tracker, onProgress) + } + if err != nil { + return nil, fmt.Errorf("fetch batch of %d files: %w", len(batch), err) + } + + byPath := make(map[string]FileBlocks, len(resp)) + for _, f := range resp { + byPath[f.RelPath] = f + } + + var written []string + for _, j := range batch { + if ctx.Err() != nil { + return written, ctx.Err() + } + f, ok := byPath[j.relPath] + if !ok { + return written, fmt.Errorf("fetch blocks for %s: %s did not send it", j.relPath, peer.Name) + } + if f.Error != "" { + return written, fmt.Errorf("fetch blocks for %s: %s", j.relPath, f.Error) + } + if err := writeBatchedFile(j, f.Blocks); err != nil { + return written, err + } + written = append(written, j.relPath) + n := changedBytes(j.remote, j.indices) + tracker.add(n) + onProgress(false) + throttle.wait(ctx, n) + } + return written, nil +} + +// writeBatchedFile writes one file from its fetched blocks, exactly as +// pullFile does: unchanged blocks from the copy on disk, the rest from the +// peer, verified and renamed into place by the PatchWriter. +func writeBatchedFile(j batchJob, blocks []BlockData) error { + writer, err := delta.NewPatchWriter(j.localPath, j.remote) + if err != nil { + return fmt.Errorf("patch %s: %w", j.relPath, err) + } + committed := false + defer func() { + if !committed { + writer.Abort() + } + }() + incoming := make(map[int]bool, len(j.indices)) + for _, idx := range j.indices { + incoming[idx] = true + } + if err := writer.SeedUnchanged(j.localPath, incoming); err != nil { + return fmt.Errorf("patch %s: %w", j.relPath, err) + } + for _, b := range blocks { + if err := writer.WriteBlock(b.Index, b.Data); err != nil { + return fmt.Errorf("patch %s: %w", j.relPath, err) + } + } + if err := writer.Commit(); err != nil { + return fmt.Errorf("patch %s: %w", j.relPath, err) + } + committed = true + if j.remote.MtimeMs > 0 { + mtime := time.UnixMilli(int64(j.remote.MtimeMs)) + _ = os.Chtimes(j.localPath, mtime, mtime) + } + return nil +} + +// pullBatchOneByOne is the fallback when the transport cannot batch for this +// peer after all: the same files, through the per-file path. +func (e *Engine) pullBatchOneByOne(ctx context.Context, peer Peer, gameID, root string, + batch []batchJob, throttle *throttler, tracker *progressTracker, onProgress func(force bool)) ([]string, error) { + + var written []string + for _, j := range batch { + if err := e.pullFile(ctx, peer, FileRef{GameID: gameID, Root: root, RelPath: j.relPath}, j.localPath, + j.remote, j.indices, throttle, tracker, onProgress); err != nil { + return written, err + } + if j.remote.MtimeMs > 0 { + mtime := time.UnixMilli(int64(j.remote.MtimeMs)) + _ = os.Chtimes(j.localPath, mtime, mtime) + } + written = append(written, j.relPath) + } + return written, nil +} diff --git a/internal/p2p/syncengine/batch_test.go b/internal/p2p/syncengine/batch_test.go new file mode 100644 index 00000000..c468cf6c --- /dev/null +++ b/internal/p2p/syncengine/batch_test.go @@ -0,0 +1,237 @@ +package syncengine + +import ( + "bytes" + "context" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + + "github.com/opensave/opensave/internal/delta" + "github.com/opensave/opensave/internal/snapshot" +) + +// batchTransport is fakeTransport plus batched fetches, answering manifests +// with a chosen protocol revision. +type batchTransport struct { + *fakeTransport + proto int + + mu sync.Mutex + batchCalls int + blockCalls int + failPath string // answered with a per-file error + // maxFiles, when set, refuses bigger batches the way a responder that + // counts differently does. + maxFiles int +} + +func (b *batchTransport) FetchManifest(ctx context.Context, peer Peer, gameID string, q ManifestQuery) (ManifestResponse, error) { + resp, err := b.fakeTransport.FetchManifest(ctx, peer, gameID, q) + resp.Proto = b.proto + return resp, err +} + +func (b *batchTransport) FetchBlocks(ctx context.Context, peer Peer, ref FileRef, idx []int, bs int) ([]BlockData, error) { + b.mu.Lock() + b.blockCalls++ + b.mu.Unlock() + return b.fakeTransport.FetchBlocks(ctx, peer, ref, idx, bs) +} + +func (b *batchTransport) FetchFileBatch(ctx context.Context, peer Peer, gameID, root string, files []FileBlocksRequest) ([]FileBlocks, error) { + b.mu.Lock() + b.batchCalls++ + b.mu.Unlock() + if len(files) > batchMaxFiles { + return nil, fmt.Errorf("batch of %d files is over the limit", len(files)) + } + if b.maxFiles > 0 && len(files) > b.maxFiles { + return nil, fmt.Errorf(`peer returned 400: {"error":"batch too large"}`) + } + out := make([]FileBlocks, 0, len(files)) + for _, f := range files { + if f.RelPath == b.failPath { + out = append(out, FileBlocks{RelPath: f.RelPath, Error: "read blocks failed: locked"}) + continue + } + blocks, err := b.fakeTransport.FetchBlocks(ctx, peer, FileRef{GameID: gameID, Root: root, RelPath: f.RelPath}, f.BlockIndices, f.BlockSize) + if err != nil { + out = append(out, FileBlocks{RelPath: f.RelPath, Error: err.Error()}) + continue + } + out = append(out, FileBlocks{RelPath: f.RelPath, Blocks: blocks}) + } + return out, nil +} + +func setupBatchEngine(t *testing.T, proto int) (*engineEnv, *batchTransport) { + t.Helper() + env := setupEngine(t) + bt := &batchTransport{fakeTransport: env.transport, proto: proto} + env.engine = New(env.store, snapshot.New(env.store), bt) + return env, bt +} + +// A save of many small files, plus one big one, in two folders. +func writeManyFiles(t *testing.T, dir string, n int) { + t.Helper() + for i := 0; i < n; i++ { + write(t, dir, fmt.Sprintf("map/chunk_%04d.bin", i), strings.Repeat(fmt.Sprintf("chunk %d;", i), 20)) + } + big := bytes.Repeat([]byte("large save data "), (512<<10)/16) // 512KB: per-file path + if err := os.WriteFile(filepath.Join(dir, "big.sav"), big, 0o666); err != nil { + t.Fatal(err) + } +} + +func assertSameTree(t *testing.T, want, got string) { + t.Helper() + wm, err := delta.BuildManifest(want) + if err != nil { + t.Fatal(err) + } + gm, err := delta.BuildManifest(got) + if err != nil { + t.Fatal(err) + } + if wm.ManifestHash() != gm.ManifestHash() { + t.Fatalf("trees differ: %d files remote, %d local", len(wm.Files), len(gm.Files)) + } + for p, f := range wm.Files { + if gm.Files[p].MtimeMs != f.MtimeMs { + t.Errorf("%s: mtime %d, want the peer's %d", p, gm.Files[p].MtimeMs, f.MtimeMs) + break + } + } +} + +func TestPull_BatchesSmallFiles(t *testing.T) { + env, bt := setupBatchEngine(t, ProtoBatchFiles) + writeManyFiles(t, env.remoteDir, 600) + + res, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer) + if err != nil { + t.Fatalf("SyncWithPeer: %v", err) + } + if res.Status != "updated" || res.Direction != "pull" { + t.Fatalf("result = %+v, want updated/pull", res) + } + assertSameTree(t, env.remoteDir, env.localDir) + + // 600 one-block files: each claims a 64 KB block as the responder counts, + // so 192 fit a request (batchMaxClaimed) -> 4 requests; the big file alone + // goes block by block. + if bt.batchCalls != 4 { + t.Errorf("batch requests = %d, want 4", bt.batchCalls) + } + if bt.blockCalls != 1 { + t.Errorf("per-file block requests = %d, want 1 (the big file)", bt.blockCalls) + } + if len(env.transport.syncEvents) == 0 || env.transport.syncEvents[len(env.transport.syncEvents)-1] != "sync-complete" { + t.Errorf("sync did not complete: %v", env.transport.syncEvents) + } +} + +// A peer that predates batching is asked file by file, as before. +func TestPull_OldPeerIsNotBatched(t *testing.T) { + env, bt := setupBatchEngine(t, ProtoMultiRoot) + writeManyFiles(t, env.remoteDir, 50) + + if _, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer); err != nil { + t.Fatalf("SyncWithPeer: %v", err) + } + assertSameTree(t, env.remoteDir, env.localDir) + if bt.batchCalls != 0 { + t.Errorf("batch requests to an old peer = %d, want 0", bt.batchCalls) + } + if bt.blockCalls != 51 { + t.Errorf("per-file requests = %d, want 51", bt.blockCalls) + } +} + +// One file the peer cannot read fails the pull and names the file, like the +// per-file path does; nothing half-written is left in its place. +func TestPull_BatchFileErrorNamesTheFile(t *testing.T) { + env, bt := setupBatchEngine(t, ProtoBatchFiles) + writeManyFiles(t, env.remoteDir, 10) + bt.failPath = "map/chunk_0003.bin" + + _, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer) + if err == nil || !strings.Contains(err.Error(), "map/chunk_0003.bin") { + t.Fatalf("err = %v, want one naming map/chunk_0003.bin", err) + } + if _, statErr := os.Stat(filepath.Join(env.localDir, "map", "chunk_0003.bin")); statErr == nil { + t.Error("the file the peer could not serve was written anyway") + } + leftovers, _ := filepath.Glob(filepath.Join(env.localDir, "map", "*"+delta.TmpSuffix)) + if len(leftovers) > 0 { + t.Errorf("temp files left behind: %v", leftovers) + } +} + +func TestGroupBatches_RespectsLimits(t *testing.T) { + var jobs []batchJob + for i := 0; i < 600; i++ { + jobs = append(jobs, batchJob{relPath: fmt.Sprint(i), + remote: delta.FileEntry{Blocks: []delta.Block{{Index: 0, Length: 32 << 10}}}, + indices: []int{0}}) + } + for _, b := range groupBatches(jobs) { + var n int64 + for _, j := range b { + n += changedBytes(j.remote, j.indices) + } + if len(b) > batchMaxFiles || n > batchMaxBytes { + t.Errorf("batch of %d files / %d bytes is over the limit", len(b), n) + } + } +} + +// Many small files, some of several blocks: what the responder counts — +// every block at full block size — must stay under its limit too. 256 files +// of one 64 KB block already claim 16 MB, and each file of more blocks tipped +// a batch over it: "batch too large", and the whole pull failed. +func TestGroupBatches_StaysUnderWhatTheResponderCounts(t *testing.T) { + var jobs []batchJob + for i := 0; i < 2000; i++ { + blocks := []delta.Block{{Index: 0, Length: 4 << 10}} + indices := []int{0} + if i%10 == 0 { + blocks = []delta.Block{{Index: 0, Length: 64 << 10}, {Index: 1, Length: 64 << 10}, {Index: 2, Length: 10 << 10}} + indices = []int{0, 1, 2} + } + jobs = append(jobs, batchJob{relPath: fmt.Sprint(i), + remote: delta.FileEntry{BlockSize: 64 << 10, Blocks: blocks}, indices: indices}) + } + for _, b := range groupBatches(jobs) { + var claimed int64 + for _, j := range b { + claimed += claimedBytes(j.remote, j.indices) + } + if claimed > 16<<20 { + t.Errorf("a batch of %d files claims %d bytes, over the responder's limit", len(b), claimed) + } + } +} + +// A responder that refuses a batch as too large gets it again in halves, +// rather than the pull failing. +func TestPull_ABatchRefusedAsTooLargeIsSplit(t *testing.T) { + env, bt := setupBatchEngine(t, ProtoBatchFiles) + bt.maxFiles = 10 + for i := 0; i < 40; i++ { + write(t, env.remoteDir, fmt.Sprintf("chunks/c%02d.bin", i), fmt.Sprint("chunk ", i)) + } + if _, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer); err != nil { + t.Fatalf("sync failed though the batches could be split: %v", err) + } + for i := 0; i < 40; i++ { + if _, err := os.Stat(filepath.Join(env.localDir, "chunks", fmt.Sprintf("c%02d.bin", i))); err != nil { + t.Fatalf("c%02d.bin did not arrive", i) + } + } +} diff --git a/internal/p2p/syncengine/engine.go b/internal/p2p/syncengine/engine.go index 5b8d8d10..53d78bab 100644 --- a/internal/p2p/syncengine/engine.go +++ b/internal/p2p/syncengine/engine.go @@ -424,15 +424,34 @@ func (e *Engine) SyncWithPeer(ctx context.Context, gameID string, peer Peer) (Re if err != nil { return Result{}, err } - e.Log("info", fmt.Sprintf("syncing %q with %q (%s)", game.Name, peer.Name, - map[bool]string{true: "WAN relay", false: "direct LAN"}[peer.Wan()])) - - // 1. Fetch remote manifest + branch info. isFile, _ := delta.ResolveLocalSaveFilePath(game.SavePath) - remoteData, err := e.Transport.FetchManifest(ctx, peer, gameID, ManifestQuery{ + query := ManifestQuery{ Name: game.Name, SavePath: game.SavePath, IsFile: isFile, AppID: game.AppID, CoverURL: game.CoverURL, - }) + } + + // 0. Nothing changed on either side since they last agreed: say so and + // stop, without moving either manifest. Quietly, too — this is the + // answer for nearly every game on every periodic pass. + res, done, prefetched := e.quickInSync(ctx, gameID, game, peer, query) + if done { + return res, nil + } + + // Said once there is something to do (after the in-sync check below), + // not here: a line per game per peer per periodic pass, most of them + // followed by "already in sync", rotated everything else out of the log + // within hours. + syncingLine := fmt.Sprintf("syncing %q with %q (%s)", game.Name, peer.Name, + map[bool]string{true: "WAN relay", false: "direct LAN"}[peer.Wan()]) + + // 1. Fetch remote manifest + branch info. + var remoteData ManifestResponse + if prefetched != nil { + remoteData = *prefetched + } else { + remoteData, err = e.Transport.FetchManifest(ctx, peer, gameID, query) + } if err != nil { // The peer simply isn't tracking this game (they untracked it, or // never had it). That's a stable state, not a transient network @@ -708,7 +727,6 @@ func (e *Engine) SyncWithPeer(ctx context.Context, gameID string, peer Peer) (Re } if !decision.HasChanges() { - e.Log("success", fmt.Sprintf("%q already in sync with %q", game.Name, peer.Name)) e.persistLineage(gameID, peer.ID, localManifest, remoteData.Manifest) // Both sides verifiably identical: this is a convergence point. _ = e.Store.SetAgreedHash(gameID, peer.ID, localManifest.ManifestHash()) @@ -727,8 +745,23 @@ func (e *Engine) SyncWithPeer(ctx context.Context, gameID string, peer Peer) (Re // are read through the gate — so the branch hold goes first. releaseAligned() e.syncExtraRoots(ctx, gameID, game, peer, remoteData) + // Share the peer's latest history entry here too. A pull does this; + // a sync that found both sides already identical used to skip it, + // so a game whose saves were identical everywhere from the start + // (Steam Cloud, a copied folder) showed a snapshot on the device + // that tracked it and none on any other. The local files ARE the + // peer's at this point, so the mirror describes the same content. + // Once per peer snapshot: recordMirrorSnapshot skips an id it has. + // + // Last, after the other locations have synced: it zips every location, + // and taking that time before them delayed their sync for nothing. + if remoteData.LatestSnapshot != nil && len(localManifest.Files) > 0 { + e.recordMirrorSnapshot(gameID, game, peer, *remoteData.LatestSnapshot, + fmt.Sprintf("Synced from peer: %s (%s)", peer.Name, remoteData.LatestSnapshot.Comment)) + } return Result{Status: "in_sync", Direction: "none"}, nil } + e.Log("info", syncingLine) if emptiedUnconfirmed(remoteData.Manifest.Files, decision, remoteData.DeletionConfirmed) { e.Log("info", fmt.Sprintf("%q holds none of %q's save files now, and has not confirmed deleting them — keeping this device's copies", @@ -1656,6 +1689,11 @@ func (e *Engine) pullFiles(ctx context.Context, peer Peer, gameID string, game s // can record the lineage from confirmed fact rather than rediscovering it // with a manifest round trip. See the sync-complete event below. var pulled []string + // Small files are collected here and fetched many to a request once the + // per-file checks below have passed; see pullBatched. + batcher, canBatch := e.Transport.(BatchFetcher) + canBatch = canBatch && !peer.Wan() && remoteData.Proto >= ProtoBatchFiles + var batched []batchJob for _, relPath := range filesToPull { if !delta.IsSafePath(root.Path, relPath) { return fmt.Errorf("path traversal attempt on pulled file %s", relPath) @@ -1703,6 +1741,10 @@ func (e *Engine) pullFiles(ctx context.Context, peer Peer, gameID string, game s if isFile, _ := delta.ResolveLocalSaveFilePath(root.Path); isFile { localFilePath = root.Path // single-file save mode } + if canBatch && len(indices) > 0 && changedBytes(remoteFile, indices) <= batchFileMaxBytes { + batched = append(batched, batchJob{relPath: relPath, localPath: localFilePath, remote: remoteFile, indices: indices}) + continue + } if err := e.pullFile(ctx, peer, FileRef{GameID: gameID, Root: root.Name, RelPath: relPath}, localFilePath, remoteFile, indices, throttle, tracker, reportProgress); err != nil { return err @@ -1721,6 +1763,13 @@ func (e *Engine) pullFiles(ctx context.Context, peer Peer, gameID string, game s // File-boundary progress reporting (always fires). reportProgress(true) } + if len(batched) > 0 { + done, err := e.pullBatched(ctx, batcher, peer, gameID, root.Name, batched, throttle, tracker, reportProgress) + pulled = append(pulled, done...) + if err != nil { + return err + } + } // Said once, plainly, at the end. A sync that quietly leaves files behind // is worse than one that fails: the save looks synced and is not, and the diff --git a/internal/p2p/syncengine/engine_test.go b/internal/p2p/syncengine/engine_test.go index aa501a7f..5827f107 100644 --- a/internal/p2p/syncengine/engine_test.go +++ b/internal/p2p/syncengine/engine_test.go @@ -230,6 +230,59 @@ func TestSync_PullNewRemoteFile(t *testing.T) { } } +// Saves identical on both sides from the start (Steam Cloud, a copied folder) +// never pull anything — and used to never share the peer's history entry +// either, so the game showed a snapshot on one device and none on the other. +func TestSync_InSyncMirrorsPeerSnapshot(t *testing.T) { + env := setupEngine(t) + write(t, env.localDir, "save.dat", "same progress") + write(t, env.remoteDir, "save.dat", "same progress") + env.transport.latestSnap = &SnapshotInfo{ID: "snap_888", Timestamp: "2026-06-01T00:00:00.000Z", Comment: "Initial snapshot"} + + res, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer) + if err != nil { + t.Fatalf("SyncWithPeer error = %v", err) + } + if res.Status != "in_sync" { + t.Fatalf("result = %+v, want in_sync", res) + } + snap, err := env.store.GetSnapshot("snap_888") + if err != nil { + t.Fatalf("peer snapshot not mirrored on an in-sync pass: %v", err) + } + if !strings.Contains(snap.Comment, "Synced from peer: Remote Device") { + t.Errorf("mirror comment = %q", snap.Comment) + } + + // A second in-sync pass must not take it again. + if _, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer); err != nil { + t.Fatal(err) + } + game, err := env.store.GetGame("game1") + if err != nil { + t.Fatal(err) + } + snaps, err := env.store.ListSnapshots("game1", game.ActiveBranch) + if err != nil { + t.Fatal(err) + } + if len(snaps) != 1 { + t.Errorf("snapshots after two in-sync passes = %d, want 1", len(snaps)) + } +} + +// An empty save folder in sync with an empty peer has nothing to mirror. +func TestSync_InSyncEmptyFolderTakesNoMirror(t *testing.T) { + env := setupEngine(t) + env.transport.latestSnap = &SnapshotInfo{ID: "snap_999", Timestamp: "2026-06-01T00:00:00.000Z", Comment: "Initial snapshot"} + if _, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer); err != nil { + t.Fatal(err) + } + if _, err := env.store.GetSnapshot("snap_999"); err == nil { + t.Error("an empty folder was mirrored") + } +} + func TestSync_InsufficientDiskSpace(t *testing.T) { env := setupEngine(t) write(t, env.remoteDir, "save.dat", "remote progress that needs space") diff --git a/internal/p2p/syncengine/partial_safety_test.go b/internal/p2p/syncengine/partial_safety_test.go index fa85845d..59588c39 100644 --- a/internal/p2p/syncengine/partial_safety_test.go +++ b/internal/p2p/syncengine/partial_safety_test.go @@ -15,7 +15,7 @@ import ( // its partial copy back, the missing files were read as "deleted over there", // and 23k files were deleted on the device that still had them. 2.4 records // as shared only what both manifests showed (persistLineage), which should -// rule that out; these hold it there. +// rule that out; these hold it there, for the per-file and the batched pull. func countFiles(t *testing.T, dir string) int { t.Helper() @@ -79,6 +79,18 @@ func TestPartialPull_InterruptedThenResumed(t *testing.T) { interruptedThenResumed(t, env, 200, 60) } +func TestPartialPull_InterruptedThenResumed_Batched(t *testing.T) { + // One batch in flight, so the first batch of 256 is written before the + // interruption lands in the second: a deterministic partial copy. + old := batchWorkers + batchWorkers = 1 + t.Cleanup(func() { batchWorkers = old }) + env, _ := setupBatchEngine(t, ProtoBatchFiles) + // Batches fetch through the same fake reader, one call per file; the + // 300th is inside the second batch. + interruptedThenResumed(t, env, 600, 300) +} + // The other side of the same situation: this device has everything, the peer // holds only part of it and never synced. The part it lacks is new to it, not // deleted by it. diff --git a/internal/p2p/syncengine/quick.go b/internal/p2p/syncengine/quick.go new file mode 100644 index 00000000..435eba66 --- /dev/null +++ b/internal/p2p/syncengine/quick.go @@ -0,0 +1,73 @@ +package syncengine + +import ( + "context" + "fmt" + + "github.com/opensave/opensave/internal/store" +) + +// quickInSync settles the common case of a sync in one small exchange: this +// device's save still hashes to the base both devices last agreed on, and +// the peer, asked whether it does too (ManifestQuery.IfHash), says +// "unchanged". Then there is nothing to decide, and neither manifest needs +// to cross the wire — for a save of a quarter-million files that is tens of +// MB of JSON per peer, which the minute-long reconcile used to fetch every +// minute to learn that nothing had changed. +// +// It only ever answers "in sync" when both sides provably hold the agreed +// state. In every other case it reports not done and the full sync runs; a +// full manifest that came back anyway (a peer that predates the question) +// is handed over in prefetched so it is not fetched twice. +// +// Deliberately narrow: +// - no exclusion rules: the agreed base is taken over the filtered +// manifest, the peer hashes its unfiltered one; +// - no extra save locations: those carry bases of their own (RootHash); +// - no conflict waiting, and the same branch on both sides. +func (e *Engine) quickInSync(ctx context.Context, gameID string, game store.Game, peer Peer, q ManifestQuery) (res Result, done bool, prefetched *ManifestResponse) { + agreed := e.Store.GetAgreedHash(gameID, peer.ID) + if agreed == "" || !e.rulesFor(gameID).Empty() { + return Result{}, false, nil + } + if roots, err := e.Store.GameRootPaths(gameID); err != nil || len(roots) > 0 { + return Result{}, false, nil + } + e.mu.Lock() + conflicted := e.activeConflicts[gameID] != nil + e.mu.Unlock() + if conflicted { + return Result{}, false, nil + } + local, err := e.ReadManifest(ctx, gameID, game.SavePath) + if err != nil || len(local.Extra) > 0 || local.ManifestHash() != agreed { + return Result{}, false, nil + } + + q.IfHash = agreed + resp, err := e.Transport.FetchManifest(ctx, peer, gameID, q) + if err != nil { + // Let the full path fetch again and classify the error as it always has. + return Result{}, false, nil + } + if !resp.Unchanged { + return Result{}, false, &resp + } + if resp.ManifestHash != agreed || (resp.ActiveBranch != "" && resp.ActiveBranch != game.ActiveBranch) { + // Unchanged, but not in a way this shortcut may act on: the full + // exchange decides. + return Result{}, false, nil + } + + // Same as the in-sync outcome of a full sync, minus what is already + // recorded: the base is unchanged and the lineage with it. + e.Transport.ReportSyncEvent(peer, gameID, "in-sync", map[string]any{ + "peerName": e.deviceName(), + "manifestHash": agreed, + }) + if resp.LatestSnapshot != nil && len(local.Files) > 0 { + e.recordMirrorSnapshot(gameID, game, peer, *resp.LatestSnapshot, + fmt.Sprintf("Synced from peer: %s (%s)", peer.Name, resp.LatestSnapshot.Comment)) + } + return Result{Status: "in_sync", Direction: "none"}, true, nil +} diff --git a/internal/p2p/syncengine/quick_test.go b/internal/p2p/syncengine/quick_test.go new file mode 100644 index 00000000..3cfdf8d8 --- /dev/null +++ b/internal/p2p/syncengine/quick_test.go @@ -0,0 +1,117 @@ +package syncengine + +import ( + "context" + "testing" + + "github.com/opensave/opensave/internal/delta" + "github.com/opensave/opensave/internal/snapshot" +) + +// ifHashTransport answers IfHash the way a current peer does, and counts +// how many full manifests it had to send. +type ifHashTransport struct { + *fakeTransport + full, unchanged int +} + +func (t *ifHashTransport) FetchManifest(ctx context.Context, peer Peer, gameID string, q ManifestQuery) (ManifestResponse, error) { + resp, err := t.fakeTransport.FetchManifest(ctx, peer, gameID, q) + if err != nil { + return resp, err + } + if q.IfHash != "" && resp.Manifest.ManifestHash() == q.IfHash { + t.unchanged++ + return ManifestResponse{ActiveBranch: resp.ActiveBranch, LatestSnapshot: resp.LatestSnapshot, + Unchanged: true, ManifestHash: q.IfHash, Manifest: delta.Manifest{}}, nil + } + t.full++ + return resp, nil +} + +func setupIfHashEngine(t *testing.T) (*engineEnv, *ifHashTransport) { + t.Helper() + env := setupEngine(t) + tr := &ifHashTransport{fakeTransport: env.transport} + env.engine = New(env.store, snapshot.New(env.store), tr) + return env, tr +} + +func TestQuickInSync_SkipsTheManifestWhenNothingChanged(t *testing.T) { + env, tr := setupIfHashEngine(t) + write(t, env.localDir, "save.dat", "same") + write(t, env.remoteDir, "save.dat", "same") + + // First sync converges and records the agreed base. + if res, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer); err != nil || res.Status != "in_sync" { + t.Fatalf("first sync = %+v, %v", res, err) + } + if tr.full != 1 { + t.Fatalf("setup: full manifests = %d, want 1", tr.full) + } + + // Nothing changed anywhere: settled without a full manifest. + res, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer) + if err != nil || res.Status != "in_sync" { + t.Fatalf("second sync = %+v, %v", res, err) + } + if tr.full != 1 || tr.unchanged != 1 { + t.Errorf("second sync fetched %d full manifests and %d unchanged answers; want 0 more full, 1 unchanged", tr.full-1, tr.unchanged) + } +} + +// A change on either side must go the full way — the shortcut may only ever +// say "in sync" when both sides provably hold the agreed state. +func TestQuickInSync_ChangesStillSync(t *testing.T) { + env, tr := setupIfHashEngine(t) + write(t, env.localDir, "save.dat", "same") + write(t, env.remoteDir, "save.dat", "same") + if _, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer); err != nil { + t.Fatal(err) + } + + // The peer changes. + write(t, env.remoteDir, "new.dat", "made on the other device") + res, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer) + if err != nil { + t.Fatal(err) + } + if res.Status != "updated" || res.Direction != "pull" { + t.Fatalf("peer change: result = %+v, want updated/pull", res) + } + + // This side changes. + write(t, env.localDir, "mine.dat", "made here") + before := env.transport.pullTriggers + res, err = env.engine.SyncWithPeer(context.Background(), "game1", env.peer) + if err != nil { + t.Fatal(err) + } + if res.Status == "in_sync" { + t.Fatalf("local change was reported in sync: %+v", res) + } + if env.transport.pullTriggers == before { + t.Error("the peer was not told to pull this side's change") + } + if tr.unchanged != 0 { + t.Errorf("unchanged answers = %d with a change on one side, want 0", tr.unchanged) + } +} + +// A peer that predates the question sends the full manifest; it is used, +// not fetched a second time. +func TestQuickInSync_OldPeerManifestIsNotFetchedTwice(t *testing.T) { + env := setupEngine(t) // plain fake: ignores IfHash + write(t, env.localDir, "save.dat", "same") + write(t, env.remoteDir, "save.dat", "same") + if _, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer); err != nil { + t.Fatal(err) + } + before := env.transport.manifestCalls + if _, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer); err != nil { + t.Fatal(err) + } + if got := env.transport.manifestCalls - before; got != 1 { + t.Errorf("manifest requests for one sync = %d, want 1", got) + } +} diff --git a/internal/p2p/syncengine/transport.go b/internal/p2p/syncengine/transport.go index c39e315c..1db8c5df 100644 --- a/internal/p2p/syncengine/transport.go +++ b/internal/p2p/syncengine/transport.go @@ -2,6 +2,7 @@ package syncengine import ( "context" + "errors" "github.com/opensave/opensave/internal/delta" ) @@ -31,6 +32,38 @@ type SnapshotInfo struct { // with more than one save location. const ProtoMultiRoot = 1 +// ProtoBatchFiles is the revision at which a peer serves the blocks of many +// files in one request (FetchFileBatch). Before it, every file cost its own +// round trip — ~200ms each in practice, which for a save of a quarter-million +// small files (Project Zomboid map chunks) meant half a day for 50MB. +const ProtoBatchFiles = 2 + +// FileBlocksRequest asks for some blocks of one file, as one entry of a batch. +type FileBlocksRequest struct { + RelPath string `json:"relPath"` + BlockIndices []int `json:"blockIndices"` + BlockSize int `json:"blockSize"` +} + +// FileBlocks is one file's answer inside a batch. Error is set, and Blocks +// empty, when the responder could not read that file. +type FileBlocks struct { + RelPath string `json:"relPath"` + Blocks []BlockData `json:"blocks"` + Error string `json:"error,omitempty"` +} + +// BatchFetcher is implemented by transports that can fetch the blocks of +// many files, all in one save location, in a single request. Optional: the +// engine checks for it, and for a peer that advertised ProtoBatchFiles. +type BatchFetcher interface { + FetchFileBatch(ctx context.Context, peer Peer, gameID, root string, files []FileBlocksRequest) ([]FileBlocks, error) +} + +// ErrBatchUnsupported is what a transport returns when it cannot batch for +// this peer (the relay path); the engine then pulls file by file. +var ErrBatchUnsupported = errors.New("batched file fetch is not supported for this peer") + // ManifestResponse is what a peer returns for a manifest request. type ManifestResponse struct { Manifest delta.Manifest `json:"manifest"` @@ -56,6 +89,13 @@ type ManifestResponse struct { // did (see hold.go). Without it an empty location is not taken as every // file in it deleted — see emptiedUnconfirmed. DeletionConfirmed bool `json:"deletionConfirmed,omitempty"` + + // Unchanged answers a request that carried ManifestQuery.IfHash: the + // responder's save still hashes to exactly that, so Manifest is left + // empty. Only ever set when asked; a peer that predates it ignores the + // question and sends the full manifest, which is handled as before. + Unchanged bool `json:"unchanged,omitempty"` + ManifestHash string `json:"manifestHash,omitempty"` } // FileRef identifies one file inside one of a game's save locations. @@ -81,6 +121,9 @@ type ManifestQuery struct { // cover art, instead of a blank tile. AppID string CoverURL string + // IfHash asks the peer to answer "unchanged" instead of its manifest when + // its save still hashes to this — the base both devices last agreed on. + IfHash string } // BlockData is one fetched block. diff --git a/internal/p2p/transport_lan.go b/internal/p2p/transport_lan.go index 11f31907..9fa1fd5c 100644 --- a/internal/p2p/transport_lan.go +++ b/internal/p2p/transport_lan.go @@ -37,6 +37,23 @@ type lanTransport struct { var lanClient = &http.Client{Timeout: 30 * time.Second} +// lanBulkClient is for the two requests whose size grows with the save: a +// manifest, and a batch of files. Thirty seconds was fine for a few hundred +// files and not for a few hundred thousand over a VPN; the manifest then never +// arrived, every sync of that game failed, and the devices never converged. +// The sync's own context still bounds it (perPeerSyncTimeout), and the +// transport still gives up on a peer that does not answer at all. +var lanBulkClient = &http.Client{ + Timeout: 10 * time.Minute, + Transport: func() http.RoundTripper { + t := http.DefaultTransport.(*http.Transport).Clone() + // Headers arrive once the peer has built its manifest, which for a big + // save not hashed recently takes a while — but not this long. + t.ResponseHeaderTimeout = 3 * time.Minute + return t + }(), +} + func peerURL(peer syncengine.Peer, route string) string { return fmt.Sprintf("http://%s:%d/api/p2p%s", peer.Address, peer.Port, route) } @@ -56,9 +73,17 @@ func (t *lanTransport) FetchManifest(ctx context.Context, peer syncengine.Peer, if q.CoverURL != "" { params.Set("coverUrl", q.CoverURL) } + if q.IfHash != "" { + params.Set("ifHash", q.IfHash) + } var resp syncengine.ManifestResponse - err := t.getJSON(ctx, peer, peerURL(peer, "/manifest/"+gameID)+"?"+params.Encode(), &resp) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, peerURL(peer, "/manifest/"+gameID)+"?"+params.Encode(), nil) + if err != nil { + return resp, err + } + t.sign(req, peer, nil) + err = doJSONWith(lanBulkClient, req, &resp) return resp, err } @@ -78,6 +103,38 @@ func (t *lanTransport) FetchBlocks(ctx context.Context, peer syncengine.Peer, re return decodeBlocks(resp.Blocks) } +// FetchFileBatch fetches the blocks of many files in one request. Only asked +// of a peer that advertised syncengine.ProtoBatchFiles. +func (t *lanTransport) FetchFileBatch(ctx context.Context, peer syncengine.Peer, gameID, root string, files []syncengine.FileBlocksRequest) ([]syncengine.FileBlocks, error) { + var resp struct { + Files []syncengine.FileBlocks `json:"files"` + } + raw, err := json.Marshal(map[string]any{"root": root, "files": files}) + if err != nil { + return nil, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, peerURL(peer, "/files/"+gameID), bytes.NewReader(raw)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + t.sign(req, peer, raw) + if err := doJSONWith(lanBulkClient, req, &resp); err != nil { + return nil, err + } + for i := range resp.Files { + if resp.Files[i].Error != "" { + continue + } + blocks, err := decodeBlocks(resp.Files[i].Blocks) + if err != nil { + return nil, fmt.Errorf("%s: %w", resp.Files[i].RelPath, err) + } + resp.Files[i].Blocks = blocks + } + return resp.Files, nil +} + func (t *lanTransport) DeleteRemote(ctx context.Context, peer syncengine.Peer, ref syncengine.FileRef) error { return t.postJSON(ctx, peer, peerURL(peer, "/delete-file/"+ref.GameID), map[string]any{"relPath": ref.RelPath, "root": ref.Root}, nil) } @@ -282,7 +339,11 @@ func (t *lanTransport) sign(req *http.Request, peer syncengine.Peer, body []byte } func doJSON(req *http.Request, out any) error { - resp, err := lanClient.Do(req) + return doJSONWith(lanClient, req, out) +} + +func doJSONWith(client *http.Client, req *http.Request, out any) error { + resp, err := client.Do(req) if err != nil { return err } diff --git a/internal/p2p/transport_wan.go b/internal/p2p/transport_wan.go index 8ed1811b..f70c6b44 100644 --- a/internal/p2p/transport_wan.go +++ b/internal/p2p/transport_wan.go @@ -154,6 +154,20 @@ func (t *routingTransport) FetchBlocks(ctx context.Context, peer syncengine.Peer return t.pick(peer).FetchBlocks(ctx, peer, ref, blockIndices, blockSize) } +// FetchFileBatch batches over the LAN only. Relay messages are sealed and +// size-limited per client (see BatchIndices), so the relay keeps pulling file +// by file. +func (t *routingTransport) FetchFileBatch(ctx context.Context, peer syncengine.Peer, gameID, root string, files []syncengine.FileBlocksRequest) ([]syncengine.FileBlocks, error) { + if peer.Wan() { + return nil, syncengine.ErrBatchUnsupported + } + b, ok := t.lan.(syncengine.BatchFetcher) + if !ok { + return nil, syncengine.ErrBatchUnsupported + } + return b.FetchFileBatch(ctx, peer, gameID, root, files) +} + func (t *routingTransport) DeleteRemote(ctx context.Context, peer syncengine.Peer, ref syncengine.FileRef) error { return t.pick(peer).DeleteRemote(ctx, peer, ref) } From 612007e73c953c821bed0096aa23afed26313bf8 Mon Sep 17 00:00:00 2001 From: ChakraFusion <112588066+ChakraFusion@users.noreply.github.com> Date: Thu, 1 Oct 2026 12:53:42 +0200 Subject: [PATCH 2/4] presence: do not trust "online" left over from the previous run Every peer started each run with whatever status the database held, and a peer counts as offline only after three failed probes. For a PC switched off overnight that meant being shown online after every start until three probes - the first ten seconds after the background loops started - had failed. - The presence loop probes at once on start, then every 10s. - The three-miss tolerance is for devices heard from during this run (answered a probe, or seen by discovery, a request or the relay since start). A peer with no sign of life since start goes offline on its first unanswered probe. Adds an opt-in e2e measurement (OPENSAVE_PRESENCE_TIMING) of how long a stopped device is shown online: 30s, by design, for one seen this run. --- e2e/presence_timing_test.go | 53 +++++++++++++++++++++++++++++++++++ internal/p2p/engine.go | 37 +++++++++++++++++++++++- internal/p2p/presence_test.go | 43 ++++++++++++++++++++++++++++ 3 files changed, 132 insertions(+), 1 deletion(-) create mode 100644 e2e/presence_timing_test.go diff --git a/e2e/presence_timing_test.go b/e2e/presence_timing_test.go new file mode 100644 index 00000000..4d9402d6 --- /dev/null +++ b/e2e/presence_timing_test.go @@ -0,0 +1,53 @@ +package e2e + +import ( + "os" + "testing" + "time" + + "github.com/opensave/opensave/testutil" +) + +// TestPresence_GoneDeviceShowsOffline measures how long a device that went +// away keeps being shown online. Opt-in measurement: +// OPENSAVE_PRESENCE_TIMING=1 go test ./e2e -run TestPresence_ -v +func TestPresence_GoneDeviceShowsOffline(t *testing.T) { + if os.Getenv("OPENSAVE_PRESENCE_TIMING") == "" { + t.Skip("set OPENSAVE_PRESENCE_TIMING to run") + } + a := testutil.NewTestDaemon(t, "Presence-A") + b := testutil.NewTestDaemon(t, "Presence-B") + a.PairWith(b) + bID := b.NodeID() + + status := func() string { + p, err := a.Daemon.Store.GetPeer(bID) + if err != nil { + return "?" + } + return p.Status + } + if !testutil.WaitFor(60*time.Second, func() bool { return status() == "online" }) { + t.Fatalf("B never showed online on A: %q", status()) + } + + gone := time.Now() + b.Server.Stop() + b.Daemon.Stop() + + last := status() + for time.Since(gone) < 10*time.Minute { + if s := status(); s != last { + t.Logf("%6.1fs after B stopped: %s -> %s", time.Since(gone).Seconds(), last, s) + last = s + } + if last == "offline" { + break + } + time.Sleep(500 * time.Millisecond) + } + t.Logf("B shown online for %.1fs after it stopped", time.Since(gone).Seconds()) + if last != "offline" { + t.Fatalf("B still %q 10 minutes after it stopped", last) + } +} diff --git a/internal/p2p/engine.go b/internal/p2p/engine.go index 203407df..f4103f7b 100644 --- a/internal/p2p/engine.go +++ b/internal/p2p/engine.go @@ -125,6 +125,10 @@ type Engine struct { pingMisses map[string]int // probeMu serializes PingPairedPeers rounds. probeMu sync.Mutex + // answered holds the peers that answered a probe during this run, and + // startedMs is when the run began (heardThisRun). + answered map[string]bool + startedMs int64 // Lineage refreshes already running after a peer-applied deletion, keyed // by game+peer. Deletions arrive one file at a time, and each one leaves @@ -293,6 +297,7 @@ func New(s *store.Store, snaps *snapshot.Manager, logf func(level, msg string)) Log: logf, ctx: ctx, cancel: cancel, + startedMs: time.Now().UnixMilli(), } e.Wan = newWanClient(e) e.RelayHost = NewRelayHost(logf) @@ -389,6 +394,26 @@ func (e *Engine) clearPingMisses(peerID string) { delete(e.pingMisses, peerID) } +// noteAnswered records that a peer answered a probe during this run. +func (e *Engine) noteAnswered(peerID string) { + e.pingMu.Lock() + defer e.pingMu.Unlock() + if e.answered == nil { + e.answered = map[string]bool{} + } + e.answered[peerID] = true +} + +// heardThisRun reports whether there is any sign of a peer since this engine +// started: a probe it answered, or a sighting (discovery, a request from it, +// the relay) recent enough to postdate the start. +func (e *Engine) heardThisRun(p store.Peer) bool { + e.pingMu.Lock() + answered := e.answered[p.ID] + e.pingMu.Unlock() + return answered || p.LastSeenMs >= e.startedMs +} + // PingPairedPeers probes every paired LAN peer and updates their // online/offline status. func (e *Engine) PingPairedPeers(ctx context.Context) { @@ -429,11 +454,18 @@ func (e *Engine) PingPairedPeers(ctx context.Context) { // failed cannot lose anything: a device that really has gone is // declared offline a few probes later, and the only cost is a sync // attempt that fails the way it would have anyway. + // + // That patience is for a device seen during this run. A status + // carried over from the previous run is not evidence of anything: a + // PC switched off overnight came back "online" from the database on + // every start, and stayed so until three probes had failed. A device + // not heard from since this start goes offline on its first miss. newStatus := p.Status if ok { e.clearPingMisses(p.ID) + e.noteAnswered(p.ID) newStatus = "online" - } else if e.notePingMiss(p.ID) >= offlineStrikes { + } else if misses := e.notePingMiss(p.ID); misses >= offlineStrikes || !e.heardThisRun(p) { newStatus = "offline" } @@ -607,6 +639,9 @@ var presenceInterval = 10 * time.Second // whatever sync work is running. It stops with the resync loop. func (e *Engine) startPresenceLoop(stop chan struct{}) { e.GoSync(func(ctx context.Context) { + // At once, then on the ticker: until the first round, every peer + // shows whatever the previous run left in the database. + e.PingPairedPeers(ctx) ticker := time.NewTicker(presenceInterval) defer ticker.Stop() for { diff --git a/internal/p2p/presence_test.go b/internal/p2p/presence_test.go index de0397de..aca90a06 100644 --- a/internal/p2p/presence_test.go +++ b/internal/p2p/presence_test.go @@ -25,6 +25,49 @@ func waitUntil(d time.Duration, cond func() bool) bool { return cond() } +// A PC switched off since the last run is in the database as "online". It must +// not stay so for the three probes a device seen this run is allowed: the +// first unanswered probe after start says it is not there. +func TestPresence_StaleOnlineFromLastRunClearsAtOnce(t *testing.T) { + old := presenceInterval + presenceInterval = time.Hour // only the probe at start + t.Cleanup(func() { presenceInterval = old }) + + s, err := store.Open(filepath.Join(t.TempDir(), "opensave.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { s.Close() }) + if err := s.EnsureDefaultSettings(t.TempDir(), t.TempDir()); err != nil { + t.Fatal(err) + } + // Nothing listens here: the PC is off. + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + port := ln.Addr().(*net.TCPAddr).Port + ln.Close() + lastRun := time.Now().Add(-10 * time.Hour).UnixMilli() + if err := s.UpsertPeer(store.Peer{ID: "node_off", Name: "Switched off", Address: "127.0.0.1", Port: port, Status: "online", LastSeenMs: lastRun}); err != nil { + t.Fatal(err) + } + + e := New(s, snapshot.New(s), func(string, string) {}) + stop := make(chan struct{}) + start := time.Now() + e.startPresenceLoop(stop) + t.Cleanup(func() { close(stop); e.cancel(); e.bgSyncWG.Wait() }) + + if !waitUntil(5*time.Second, func() bool { + p, err := s.GetPeer("node_off") + return err == nil && p.Status == "offline" + }) { + t.Fatal("a peer online only in the last run's record was still online after the first probe") + } + t.Logf("shown offline %.1fs after start", time.Since(start).Seconds()) +} + // A peer's online state follows it on its own ticker, whatever else the // engine is doing — not only when a sync or the minute-long reconcile happens // to probe. A peer reached over a VPN gets no UDP broadcasts, so without this From 13cbf650649d84bb5e3a699ad1e7853a1ba48791 Mon Sep 17 00:00:00 2001 From: ChakraFusion <112588066+ChakraFusion@users.noreply.github.com> Date: Sat, 17 Oct 2026 18:02:25 +0200 Subject: [PATCH 3/4] sync: a batched pull stays OpenSave's own write when given the peer's time The patch writer settles a pulled file as OpenSave's own when it commits it (owntouch, #23). The batched pull then gave the file the peer's modification time without marking it again, as the per-file path does, so the exact time-and-size comparison failed and the watcher took the pull for the game saving: an automatic snapshot after every sync of small files, and a sync back of a change this device never made. Found once #52's test ran against this branch on dev-public (TestLauncherFile_SteamRewritingRemotecacheIsNoSaveChange: 4/4 failures before, 10/10 passes after, in separate runs). Both batched paths now set the time the way the per-file path does (setPulledMtime: mark, set, settle). Test: TestSetPulledMtime_KeepsThePullOpenSavesOwn fails without the marking. Co-Authored-By: Claude Opus 5.5 --- internal/p2p/syncengine/batch.go | 26 ++++++++++++++++++-------- internal/p2p/syncengine/batch_test.go | 22 ++++++++++++++++++++++ 2 files changed, 40 insertions(+), 8 deletions(-) diff --git a/internal/p2p/syncengine/batch.go b/internal/p2p/syncengine/batch.go index 7a15d94e..4e7c6821 100644 --- a/internal/p2p/syncengine/batch.go +++ b/internal/p2p/syncengine/batch.go @@ -10,6 +10,7 @@ import ( "time" "github.com/opensave/opensave/internal/delta" + "github.com/opensave/opensave/internal/owntouch" ) // Pulling many small files, many to a request. @@ -268,13 +269,25 @@ func writeBatchedFile(j batchJob, blocks []BlockData) error { return fmt.Errorf("patch %s: %w", j.relPath, err) } committed = true - if j.remote.MtimeMs > 0 { - mtime := time.UnixMilli(int64(j.remote.MtimeMs)) - _ = os.Chtimes(j.localPath, mtime, mtime) - } + setPulledMtime(j.localPath, j.remote.MtimeMs) return nil } +// setPulledMtime gives a pulled file the peer's modification time, as the +// per-file path does: marked again first, and settled after, so the time it +// ends with is known as OpenSave's own (owntouch). Set without that, the +// watcher took a batched pull for the game saving, and snapshotted and synced +// it back as a change of this device's own. +func setPulledMtime(path string, mtimeMs delta.Milli) { + if mtimeMs <= 0 { + return + } + mtime := time.UnixMilli(int64(mtimeMs)) + owntouch.Mark(path) + _ = os.Chtimes(path, mtime, mtime) + owntouch.Settled(path) +} + // pullBatchOneByOne is the fallback when the transport cannot batch for this // peer after all: the same files, through the per-file path. func (e *Engine) pullBatchOneByOne(ctx context.Context, peer Peer, gameID, root string, @@ -286,10 +299,7 @@ func (e *Engine) pullBatchOneByOne(ctx context.Context, peer Peer, gameID, root j.remote, j.indices, throttle, tracker, onProgress); err != nil { return written, err } - if j.remote.MtimeMs > 0 { - mtime := time.UnixMilli(int64(j.remote.MtimeMs)) - _ = os.Chtimes(j.localPath, mtime, mtime) - } + setPulledMtime(j.localPath, j.remote.MtimeMs) written = append(written, j.relPath) } return written, nil diff --git a/internal/p2p/syncengine/batch_test.go b/internal/p2p/syncengine/batch_test.go index c468cf6c..18984e2d 100644 --- a/internal/p2p/syncengine/batch_test.go +++ b/internal/p2p/syncengine/batch_test.go @@ -9,8 +9,10 @@ import ( "strings" "sync" "testing" + "time" "github.com/opensave/opensave/internal/delta" + "github.com/opensave/opensave/internal/owntouch" "github.com/opensave/opensave/internal/snapshot" ) @@ -235,3 +237,23 @@ func TestPull_ABatchRefusedAsTooLargeIsSplit(t *testing.T) { } } } + +// A batched pull gives each file the peer's modification time after the patch +// writer has committed it, and with it settled it as OpenSave's own +// (owntouch). Changing the time without marking it again made the file the +// game's: the watcher snapshotted the pull and synced it back as a change of +// this device's own (TestLauncherFile_SteamRewritingRemotecacheIsNoSaveChange +// on #27's CI). +func TestSetPulledMtime_KeepsThePullOpenSavesOwn(t *testing.T) { + path := filepath.Join(t.TempDir(), "save.dat") + if err := os.WriteFile(path, []byte("pulled"), 0o666); err != nil { + t.Fatal(err) + } + owntouch.Mark(path) // as the patch writer's commit leaves it + owntouch.Settled(path) + + setPulledMtime(path, delta.Milli(time.Now().Add(-time.Hour).UnixMilli())) + if !owntouch.Recent(path) { + t.Error("a pulled file given the peer's time is no longer known as OpenSave's own") + } +} From 527f3eacb0722beb7846f1b4d9053f23ac28d481 Mon Sep 17 00:00:00 2001 From: ChakraFusion <112588066+ChakraFusion@users.noreply.github.com> Date: Sat, 17 Oct 2026 20:52:15 +0200 Subject: [PATCH 4/4] sync: an in-sync pass mirrors the peer's snapshot only where none is kept yet An in-sync pass shares the peer's latest snapshot so a game identical everywhere from the start (Steam Cloud, a copied folder) has history on every device. It mirrored it whenever the peer's latest snapshot was newer, also on a device that already keeps a snapshot of its save as it stands: a second copy of the same save, named as synced from a peer that sent nothing. Even the peer's first snapshot of an empty folder was taken this way. Found by TestLauncherFile_SteamRewritingRemotecacheIsNoSaveChange, which failed or passed depending on which device tracked the game a few milliseconds later (always on Linux and Windows CI for #27). The mirror is now taken only where this device keeps no snapshot of the current save yet: the content recorded with its last snapshot (the watcher's baseline) is not the content now (keepsCurrentSave). Same in the quick "unchanged" path. Test: TestSync_InSyncDoesNotMirrorASaveAlreadyKept fails without the check; TestSync_InSyncMirrorsPeerSnapshot still passes. Co-Authored-By: Claude Opus 5.5 --- internal/p2p/syncengine/engine.go | 21 ++++++++++++++++++- internal/p2p/syncengine/engine_test.go | 29 ++++++++++++++++++++++++++ internal/p2p/syncengine/quick.go | 2 +- 3 files changed, 50 insertions(+), 2 deletions(-) diff --git a/internal/p2p/syncengine/engine.go b/internal/p2p/syncengine/engine.go index 53d78bab..77c3b8ba 100644 --- a/internal/p2p/syncengine/engine.go +++ b/internal/p2p/syncengine/engine.go @@ -752,10 +752,12 @@ func (e *Engine) SyncWithPeer(ctx context.Context, gameID string, peer Peer) (Re // that tracked it and none on any other. The local files ARE the // peer's at this point, so the mirror describes the same content. // Once per peer snapshot: recordMirrorSnapshot skips an id it has. + // And only where this device keeps no snapshot of this save yet + // (keepsCurrentSave): nothing came from the peer in this pass. // // Last, after the other locations have synced: it zips every location, // and taking that time before them delayed their sync for nothing. - if remoteData.LatestSnapshot != nil && len(localManifest.Files) > 0 { + if remoteData.LatestSnapshot != nil && len(localManifest.Files) > 0 && !e.keepsCurrentSave(gameID, game, localManifest) { e.recordMirrorSnapshot(gameID, game, peer, *remoteData.LatestSnapshot, fmt.Sprintf("Synced from peer: %s (%s)", peer.Name, remoteData.LatestSnapshot.Comment)) } @@ -1970,6 +1972,23 @@ func fetchWithRetry(ctx context.Context, t Transport, peer Peer, ref FileRef, return nil, lastErr } +// keepsCurrentSave reports whether this device already has a snapshot of its +// save as it stands: the content recorded with its last snapshot (the +// watcher's baseline) is the content now. A sync that found both sides +// identical then has nothing to add to the history. Mirroring the peer's +// latest snapshot anyway made a second copy of a save already kept, named +// as synced from a peer that sent nothing — whenever the peer's latest +// snapshot happened to be newer, even its first one of an empty folder +// (TestLauncherFile_SteamRewritingRemotecacheIsNoSaveChange, which then +// failed or passed by which device had tracked the game a few milliseconds +// later). +func (e *Engine) keepsCurrentSave(gameID string, game store.Game, local delta.Manifest) bool { + if g, err := e.Store.GetGame(gameID); err == nil { + game = g + } + return game.LastManifestHash != "" && game.LastManifestHash == e.contentHashOf(gameID, local, game.SavePath) +} + // recordMirrorSnapshot zips the (just-updated) local save under the peer's // snapshot id so both devices show the same history entry. func (e *Engine) recordMirrorSnapshot(gameID string, game store.Game, peer Peer, remoteSnap SnapshotInfo, comment string) { diff --git a/internal/p2p/syncengine/engine_test.go b/internal/p2p/syncengine/engine_test.go index 5827f107..8122b9a4 100644 --- a/internal/p2p/syncengine/engine_test.go +++ b/internal/p2p/syncengine/engine_test.go @@ -271,6 +271,35 @@ func TestSync_InSyncMirrorsPeerSnapshot(t *testing.T) { } } +// A device that already keeps a snapshot of its save as it stands takes no +// mirror on an in-sync pass: nothing came from the peer, and the mirror would +// be a second copy of the same save, named as synced from the peer. It used to +// be taken whenever the peer's latest snapshot was newer. +func TestSync_InSyncDoesNotMirrorASaveAlreadyKept(t *testing.T) { + env := setupEngine(t) + write(t, env.localDir, "save.dat", "same progress") + write(t, env.remoteDir, "save.dat", "same progress") + if _, err := env.engine.Snapshots.Create("game1", "Initial snapshot", true); err != nil { + t.Fatal(err) + } + m, err := delta.BuildManifest(env.localDir) + if err != nil { + t.Fatal(err) + } + // As the watcher records it after that snapshot. + if err := env.store.SetLastManifestHash("game1", env.engine.contentHashOf("game1", m, env.localDir)); err != nil { + t.Fatal(err) + } + env.transport.latestSnap = &SnapshotInfo{ID: "snap_99999999999999", Timestamp: "2099-01-01T00:00:00.000Z", Comment: "Initial snapshot"} + + if res, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer); err != nil || res.Status != "in_sync" { + t.Fatalf("result = %+v, %v; want in_sync", res, err) + } + if _, err := env.store.GetSnapshot("snap_99999999999999"); err == nil { + t.Error("the peer's snapshot was mirrored over a save this device already keeps") + } +} + // An empty save folder in sync with an empty peer has nothing to mirror. func TestSync_InSyncEmptyFolderTakesNoMirror(t *testing.T) { env := setupEngine(t) diff --git a/internal/p2p/syncengine/quick.go b/internal/p2p/syncengine/quick.go index 435eba66..fa07d6e1 100644 --- a/internal/p2p/syncengine/quick.go +++ b/internal/p2p/syncengine/quick.go @@ -65,7 +65,7 @@ func (e *Engine) quickInSync(ctx context.Context, gameID string, game store.Game "peerName": e.deviceName(), "manifestHash": agreed, }) - if resp.LatestSnapshot != nil && len(local.Files) > 0 { + if resp.LatestSnapshot != nil && len(local.Files) > 0 && !e.keepsCurrentSave(gameID, game, local) { e.recordMirrorSnapshot(gameID, game, peer, *resp.LatestSnapshot, fmt.Sprintf("Synced from peer: %s (%s)", peer.Name, resp.LatestSnapshot.Comment)) }