From 077cc93599f1747baf0ecd81cf961806df971bfe 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 01/10] 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 | 60 +++- 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 | 42 +++ internal/p2p/transport_lan.go | 65 +++- internal/p2p/transport_wan.go | 14 + 16 files changed, 1407 insertions(+), 11 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 ba2142ff..5776c394 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) }) @@ -732,7 +735,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 @@ -795,6 +810,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 { @@ -1003,6 +1112,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) @@ -1043,7 +1178,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 3a0089fa..256ad75d 100644 --- a/internal/p2p/syncengine/engine.go +++ b/internal/p2p/syncengine/engine.go @@ -434,16 +434,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) query := ManifestQuery{ Name: game.Name, SavePath: game.SavePath, IsFile: isFile, AppID: game.AppID, CoverURL: game.CoverURL, } - remoteData, err := e.Transport.FetchManifest(ctx, peer, gameID, query) + + // 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 @@ -761,7 +779,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()) @@ -780,8 +797,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", @@ -1711,6 +1743,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) @@ -1758,6 +1795,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 @@ -1776,6 +1817,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 d7ab8bc9..e9814ad4 100644 --- a/internal/p2p/syncengine/engine_test.go +++ b/internal/p2p/syncengine/engine_test.go @@ -233,6 +233,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 aeb5d77b..93e54562 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"` @@ -60,6 +93,12 @@ type ManifestResponse struct { // HeldYours answers ManifestQuery.Held: a snapshot here holds exactly the // asker's files (heldstate.go). Absent from a peer that predates it. HeldYours bool `json:"heldYours,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. @@ -88,6 +127,9 @@ type ManifestQuery struct { // Held identifies this device's files (heldstate.go heldKey), sent only when it // has no agreed state with the peer: has the peer held this save? Held 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 09040cae..7fae41da 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) } @@ -59,9 +76,17 @@ func (t *lanTransport) FetchManifest(ctx context.Context, peer syncengine.Peer, if q.Held != "" { params.Set("held", q.Held) } + 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 } @@ -81,6 +106,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) } @@ -285,7 +342,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 93172dd2..543f5f79 100644 --- a/internal/p2p/transport_wan.go +++ b/internal/p2p/transport_wan.go @@ -155,6 +155,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 74e7c32ab0a3fd018c94adb9daf44a539e2383d3 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 02/10] 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 bb08cbc4d8a353f44e6d775b7cf3b418faf46af0 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 03/10] 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 ba677ebc5ac254f2b2d2ce0ce42a80bfa3acc4b3 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 04/10] 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 256ad75d..867c23f0 100644 --- a/internal/p2p/syncengine/engine.go +++ b/internal/p2p/syncengine/engine.go @@ -804,10 +804,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)) } @@ -2024,6 +2026,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 e9814ad4..9835fa9a 100644 --- a/internal/p2p/syncengine/engine_test.go +++ b/internal/p2p/syncengine/engine_test.go @@ -274,6 +274,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)) } From 07812900bde3121f73e4405fa181910afc2e8104 Mon Sep 17 00:00:00 2001 From: ChakraFusion <112588066+ChakraFusion@users.noreply.github.com> Date: Fri, 2 Oct 2026 12:50:18 +0200 Subject: [PATCH 05/10] sync: closest devices first, and no fixed 30-second limit on block fetches Network-aware order: - Each device records how fast its connection to every other one is: from the pulls it makes, and from a short speed test (/api/p2p/speedtest, 2 MB, at most every 6 hours) when nothing has measured it. Before anything is measured, the address kind stands in: home network fast, VPN (Tailscale's range) slower, the internet relay slowest. - A sync goes to the fastest devices first. A device that is behind takes the newer save from the fastest device holding it. - Having asked a close device to take a newer save, a sync waits for it to finish (it reports sync-complete) before asking a much slower one, bounded by how long the transfer should take. The save reaches the close devices at full speed, and the slow ones can then take it from whichever device is nearest to them. - The device list shows each connection's kind and measured speed. Block fetches on direct addresses had a fixed 30-second limit. "Direct" includes VPN addresses, and a few MB of blocks at a VPN relay's hundred-odd KB a second outlasted it: the request failed, and with it the whole pull, over and over. The limit now grows with what is asked for, as it already did over the internet relay. Co-Authored-By: Claude Opus 5.5 --- .../frontend/src/views/Devices.svelte | 14 + internal/api/server.go | 7 +- internal/p2p/engine.go | 34 +++ internal/p2p/routes.go | 36 +++ internal/p2p/syncengine/engine.go | 35 ++- internal/p2p/syncengine/linkspeed.go | 273 ++++++++++++++++++ internal/p2p/syncengine/linkspeed_test.go | 106 +++++++ internal/p2p/transport_lan.go | 48 ++- internal/p2p/transport_wan.go | 11 + internal/p2p/wanclient_handlers.go | 1 + internal/store/migrations/0038_peer_links.sql | 15 + internal/store/peer_links.go | 37 +++ 12 files changed, 613 insertions(+), 4 deletions(-) create mode 100644 internal/p2p/syncengine/linkspeed.go create mode 100644 internal/p2p/syncengine/linkspeed_test.go create mode 100644 internal/store/migrations/0038_peer_links.sql create mode 100644 internal/store/peer_links.go diff --git a/cmd/opensave-app/frontend/src/views/Devices.svelte b/cmd/opensave-app/frontend/src/views/Devices.svelte index 641c5f3b..4b53ad09 100644 --- a/cmd/opensave-app/frontend/src/views/Devices.svelte +++ b/cmd/opensave-app/frontend/src/views/Devices.svelte @@ -76,6 +76,15 @@ }; const fmtTime = (t) => (t ? new Date(t).toLocaleString() : 'never'); + + // How this device reaches the peer, and how fast it has found that to be. + const linkKinds = { lan: 'home network', vpn: 'VPN (e.g. Tailscale)', relay: 'internet relay', internet: 'internet' }; + const fmtRate = (b) => (b >= 1048576 ? (b / 1048576).toFixed(1) + ' MB/s' : Math.max(1, Math.round(b / 1024)) + ' KB/s'); + function linkLabel(link) { + const kind = linkKinds[link.kind] ?? link.kind; + if (!link.bytesPerSec) return `${kind} · speed not measured yet`; + return `${kind} · ${fmtRate(link.bytesPerSec)} (measured ${fmtTime(link.measuredMs)})`; + }
@@ -105,6 +114,11 @@ · last synced {fmtTime(peer.lastSynced)} {#if peer.appVersion}· OpenSave {peer.appVersion}{/if}
+ {#if peer.link} +
+ {linkLabel(peer.link)} +
+ {/if}
diff --git a/internal/api/server.go b/internal/api/server.go index 4d40a280..dbc8ffb7 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -101,12 +101,17 @@ func (s *Server) peersPayload() map[string]any { AppVersion string `json:"appVersion,omitempty"` BuildTimeMs int64 `json:"buildTimeMs,omitempty"` HasNewerBuild bool `json:"hasNewerBuild,omitempty"` + // Link is how fast this device has found its connection to it, + // and what kind of address it is reached on: what decides the + // order devices are synced in. + Link syncengine.LinkStat `json:"link"` // What actually protects traffic with this device. Sent per peer // rather than described once in the interface, because the answer // differs per pairing and the reader cannot work out which case // they are in from a general statement. p2p.PeerProtection - }{Peer: p, PeerProtection: s.Daemon.P2P.PeerProtection(p)} + }{Peer: p, PeerProtection: s.Daemon.P2P.PeerProtection(p), + Link: s.Daemon.P2P.Sync.Link(syncengine.Peer{ID: p.ID, Name: p.Name, Address: p.Address, Port: p.Port, IsWan: p.Address == "relay"})} if b, ok := builds[p.ID]; ok { entry.AppVersion = b.AppVersion entry.BuildTimeMs = b.BuildTimeMs diff --git a/internal/p2p/engine.go b/internal/p2p/engine.go index f4103f7b..8c9229bb 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 + // probingLinks: peers whose connection is being speed-tested now + // (probeLinkSoon). + probeLinksMu sync.Mutex + probingLinks map[string]bool // answered holds the peers that answered a probe during this run, and // startedMs is when the run began (heardThisRun). answered map[string]bool @@ -465,6 +469,7 @@ func (e *Engine) PingPairedPeers(ctx context.Context) { e.clearPingMisses(p.ID) e.noteAnswered(p.ID) newStatus = "online" + e.probeLinkSoon(p) } else if misses := e.notePingMiss(p.ID); misses >= offlineStrikes || !e.heardThisRun(p) { newStatus = "offline" } @@ -481,6 +486,35 @@ func (e *Engine) PingPairedPeers(ctx context.Context) { } } +// probeLinkSoon tests the connection to a peer that just answered, in the +// background, when nothing has measured it for a while +// (syncengine/linkspeed.go). One test per peer at a time. +func (e *Engine) probeLinkSoon(p store.Peer) { + if e.Sync == nil || p.Address == "relay" { + return + } + e.probeLinksMu.Lock() + if e.probingLinks == nil { + e.probingLinks = map[string]bool{} + } + if e.probingLinks[p.ID] { + e.probeLinksMu.Unlock() + return + } + e.probingLinks[p.ID] = true + e.probeLinksMu.Unlock() + go func() { + defer func() { + e.probeLinksMu.Lock() + delete(e.probingLinks, p.ID) + e.probeLinksMu.Unlock() + }() + if e.Sync.ProbeLinkIfDue(context.Background(), syncengine.Peer{ID: p.ID, Name: p.Name, Address: p.Address, Port: p.Port}) { + e.notifyPeerUpdate() + } + }() +} + // ErrNoPeersOnline is a sync with no other device to sync with: none of the // paired devices answered. Not a failure of anything — the save goes over when // one is back — which is why it is told apart from errors that are. diff --git a/internal/p2p/routes.go b/internal/p2p/routes.go index 5776c394..9a6e2152 100644 --- a/internal/p2p/routes.go +++ b/internal/p2p/routes.go @@ -3,6 +3,7 @@ package p2p import ( "compress/gzip" "context" + "crypto/rand" "encoding/json" "errors" "fmt" @@ -43,6 +44,7 @@ func (e *Engine) RegisterRoutes(r chi.Router) { r.Get("/api/p2p/games", e.handlePeerGameList) r.Post("/api/p2p/sync-event/{gameId}", e.handleSyncEvent) r.Get("/api/p2p/app-binary", e.handleAppBinary) + r.Get("/api/p2p/speedtest", e.handleSpeedTest) // What moves save data, or starts a sync: refused while paused. r.Group(func(r chi.Router) { @@ -56,6 +58,34 @@ func (e *Engine) RegisterRoutes(r chi.Router) { }) } +// speedTestMax bounds one speed test, and speedTestData is what it sends: +// random, so nothing along the way can compress it into a better figure. +const speedTestMax = 8 << 20 + +var speedTestData = func() []byte { + b := make([]byte, 1<<20) + _, _ = rand.Read(b) + return b +}() + +// handleSpeedTest sends n bytes (at most speedTestMax) for a peer to time: +// how it learns how close this device is (syncengine/linkspeed.go). +func (e *Engine) handleSpeedTest(w http.ResponseWriter, r *http.Request) { + n, _ := strconv.Atoi(r.URL.Query().Get("n")) + if n <= 0 || n > speedTestMax { + n = speedTestMax + } + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Content-Length", strconv.Itoa(n)) + for n > 0 { + chunk := min(n, len(speedTestData)) + if _, err := w.Write(speedTestData[:chunk]); err != nil { + return + } + n -= chunk + } +} + func clientIP(r *http.Request) string { host, _, err := net.SplitHostPort(r.RemoteAddr) if err != nil { @@ -1071,6 +1101,10 @@ func (e *Engine) handleSyncEvent(w http.ResponseWriter, r *http.Request) { if e.Sync.Progress.OnSyncError != nil { e.Sync.Progress.OnSyncError(gameID, ev) } + // A sync here waiting for this peer to finish need not wait on. + if peer, ok := e.peerByAddress(clientIP(r)); ok { + e.Sync.NotePeerPulled(gameID, peer.ID) + } } jsonOK(w, map[string]any{"success": true}) } @@ -1199,6 +1233,8 @@ func SetServedProto(v int) int { // freshly-pushed files deliberately stay out of it (see persistLineage), so // deleting one here would pull it back instead of propagating the delete. func (e *Engine) peerFinishedPulling(gameID string, peer syncengine.Peer, data map[string]any) { + // A sync here waiting for this peer before slower ones goes on. + defer e.Sync.NotePeerPulled(gameID, peer.ID) // Recorded first, and synchronously: the peer said exactly which files it // wrote, so the lineage can be updated now rather than after a manifest // round trip. That round trip is what left a window in which deleting a diff --git a/internal/p2p/syncengine/engine.go b/internal/p2p/syncengine/engine.go index 867c23f0..19040386 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 + + // linkMu guards links, the measured connection to each peer, and + // pullWaits, syncs waiting for a close peer to finish taking a save + // (linkspeed.go). + linkMu sync.Mutex + links map[string]store.PeerLink + pullWaits map[string]chan struct{} } // New creates an Engine. @@ -331,13 +338,32 @@ func (e *Engine) SyncGame(ctx context.Context, gameID string, onlinePeers []Peer results := map[string]Result{} busy := false - for _, peer := range onlinePeers { + // Fastest connections first (linkspeed.go): a device that is behind takes + // the newer save from the closest device holding it, and a newer save + // here reaches the close devices before the slow ones. + onlinePeers = e.OrderBySpeed(onlinePeers) + for i, peer := range onlinePeers { // Hard per-peer cap: a wedged transport must never hold // activeSyncs forever (which would silently block every future // sync of this game until an app restart). peerCtx, cancel := context.WithTimeout(ctx, perPeerSyncTimeout) res, err := e.SyncWithPeer(peerCtx, gameID, peer) cancel() + // A close device asked to take this save is given the time to, before + // a much slower one is asked: it gets it at full speed, and the slow + // one can then take it from whichever device is nearest to it. + if err == nil && res.Status == "triggered_peer_pull" && e.waitsForClose(onlinePeers, i) { + var size int64 + if game, gerr := e.Store.GetGame(gameID); gerr == nil { + if m, merr := e.ReadManifest(ctx, gameID, game.SavePath); merr == nil { + for _, f := range m.Files { + size += f.Size + } + } + } + e.Log("info", fmt.Sprintf("waiting for %s (close by) to take %s before the slower devices", peer.Name, gameID)) + e.waitForPull(ctx, gameID, peer, size) + } if err == nil && res.Status == "peer_busy" { // Nothing was compared, so nothing is stamped, and it is not a // failure: the peer was writing this game for a sync of its own. @@ -1707,6 +1733,13 @@ func (e *Engine) pullFiles(ctx context.Context, peer Peer, gameID string, game s tracker := newProgressTracker(totalBytes) throttle := e.throttleFor(peer.Wan()) + // What this transfer says about the connection (linkspeed.go), measured + // however it ends: a pull cut off half-way still moved what it moved. + pullStarted := time.Now() + defer func() { + moved, _, _ := tracker.stats() + e.noteLinkRate(peer, moved, time.Since(pullStarted)) + }() // Progress reporter shared by the per-file loop and the block-group // loop inside each file. Without in-file reporting, a single large diff --git a/internal/p2p/syncengine/linkspeed.go b/internal/p2p/syncengine/linkspeed.go new file mode 100644 index 00000000..b19543b1 --- /dev/null +++ b/internal/p2p/syncengine/linkspeed.go @@ -0,0 +1,273 @@ +package syncengine + +import ( + "context" + "fmt" + "net" + "sort" + "time" + + "github.com/opensave/opensave/internal/store" +) + +// Network-aware syncing. +// +// Devices are not equally close. Two on the same home network move a save at +// tens of MB a second; one reached over a VPN relay or the internet relay may +// manage a hundred KB. A save that goes to every device at once spends hours +// crawling over the slow links while the device next door, which could have +// had it in seconds and then handed it on, waits in the same queue. +// +// So each device keeps how fast it has found its connection to every other — +// measured from the transfers it makes, and from a short speed test when it +// has nothing recent — and works fastest first: a sync goes to the close +// devices first, and a device that is behind takes the newer save from the +// fastest device that holds it. Before anything is measured, the kind of +// address stands in: a home-network address is taken for fast, a VPN address +// (Tailscale's range) for slower, the internet relay for slowest. + +// LinkStat is what this device knows about its connection to a peer. +type LinkStat struct { + // BytesPerSec is the measured rate, 0 when nothing is measured yet. + BytesPerSec float64 `json:"bytesPerSec"` + MeasuredMs int64 `json:"measuredMs"` + // Kind is what the address says: "lan", "vpn", "relay" or "internet". + Kind string `json:"kind"` +} + +const ( + // minMeasureBytes and minMeasureTime: smaller transfers say more about + // round trips than about the link, and are not counted. + minMeasureBytes = 512 << 10 + minMeasureTime = 500 * time.Millisecond + // probeEvery is how often a link is tested again when no transfer has + // measured it since; probeBytes how much a test moves. + probeEvery = 6 * time.Hour + probeBytes = 2 << 20 +) + +// prior rates, by address kind, used until a link is measured. +var priorRate = map[string]float64{ + "lan": 40 << 20, + "vpn": 2 << 20, + "internet": 1 << 20, + "relay": 256 << 10, +} + +var ( + _, tailscaleNet, _ = net.ParseCIDR("100.64.0.0/10") + privateNets = func() []*net.IPNet { + var out []*net.IPNet + for _, c := range []string{"10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "169.254.0.0/16", "fc00::/7", "fe80::/10"} { + _, n, _ := net.ParseCIDR(c) + out = append(out, n) + } + return out + }() +) + +// linkKind says what a peer's address is. +func linkKind(peer Peer) string { + if peer.Wan() { + return "relay" + } + ip := net.ParseIP(peer.Address) + if ip == nil { + return "internet" + } + if ip.IsLoopback() { + return "lan" + } + if tailscaleNet.Contains(ip) { + return "vpn" + } + for _, n := range privateNets { + if n.Contains(ip) { + return "lan" + } + } + return "internet" +} + +func (e *Engine) loadLinksLocked() { + if e.links != nil { + return + } + e.links = map[string]store.PeerLink{} + if e.Store == nil { + return + } + if links, err := e.Store.PeerLinks(); err == nil { + e.links = links + } +} + +// Link is what this device knows about its connection to a peer. +func (e *Engine) Link(peer Peer) LinkStat { + e.linkMu.Lock() + defer e.linkMu.Unlock() + e.loadLinksLocked() + l := e.links[peer.ID] + return LinkStat{BytesPerSec: l.BytesPerSec, MeasuredMs: l.MeasuredMs, Kind: linkKind(peer)} +} + +// expectedRate is how fast a transfer with a peer is expected to go: what was +// measured, or the address kind's guess. +func (e *Engine) expectedRate(peer Peer) float64 { + l := e.Link(peer) + if l.BytesPerSec > 0 { + return l.BytesPerSec + } + return priorRate[l.Kind] +} + +// noteLinkRate records a transfer with a peer: bytes in d. +func (e *Engine) noteLinkRate(peer Peer, bytes int64, d time.Duration) { + if bytes < minMeasureBytes || d < minMeasureTime { + return + } + rate := float64(bytes) / d.Seconds() + e.linkMu.Lock() + defer e.linkMu.Unlock() + e.loadLinksLocked() + l := e.links[peer.ID] + l.PeerID = peer.ID + if l.BytesPerSec > 0 { + // A running average, weighted to the latest: a link changes (a VPN + // that found a direct path, a laptop that moved networks), and the + // order should follow it within a transfer or two. + l.BytesPerSec = 0.4*l.BytesPerSec + 0.6*rate + } else { + l.BytesPerSec = rate + } + l.MeasuredMs = time.Now().UnixMilli() + e.links[peer.ID] = l + if e.Store != nil { + _ = e.Store.SavePeerLink(l) + } +} + +// OrderBySpeed sorts peers fastest first, keeping the given order among +// equals. +func (e *Engine) OrderBySpeed(peers []Peer) []Peer { + out := append([]Peer(nil), peers...) + rates := make(map[string]float64, len(out)) + for _, p := range out { + rates[p.ID] = e.expectedRate(p) + } + sort.SliceStable(out, func(i, j int) bool { return rates[out[i].ID] > rates[out[j].ID] }) + return out +} + +// SpeedProber is implemented by transports that can test a link: fetch n +// bytes of nothing in particular from the peer and say how long it took. +type SpeedProber interface { + ProbeSpeed(ctx context.Context, peer Peer, n int) (int64, time.Duration, error) +} + +// ProbeLinkIfDue tests the link to a peer when nothing has measured it for a +// while. Not over the internet relay, which is the slowest link whatever a +// test says and would only be loaded further. Reports whether it ran. +func (e *Engine) ProbeLinkIfDue(ctx context.Context, peer Peer) bool { + prober, ok := e.Transport.(SpeedProber) + if !ok || peer.Wan() { + return false + } + e.linkMu.Lock() + e.loadLinksLocked() + l := e.links[peer.ID] + now := time.Now() + due := now.Sub(time.UnixMilli(l.MeasuredMs)) > probeEvery && now.Sub(time.UnixMilli(l.ProbedMs)) > probeEvery + if due { + l.PeerID = peer.ID + l.ProbedMs = now.UnixMilli() + e.links[peer.ID] = l + if e.Store != nil { + _ = e.Store.SavePeerLink(l) + } + } + e.linkMu.Unlock() + if !due { + return false + } + ctx, cancel := context.WithTimeout(ctx, 60*time.Second) + defer cancel() + n, d, err := prober.ProbeSpeed(ctx, peer, probeBytes) + if err != nil { + e.Log("info", fmt.Sprintf("speed test with %s did not finish: %v", peer.Name, err)) + return true + } + e.noteLinkRate(peer, n, d) + e.Log("info", fmt.Sprintf("connection to %s: %s/s (%s)", peer.Name, humanBytes(int64(e.Link(peer).BytesPerSec)), linkKind(peer))) + return true +} + +// Waiting for close devices before slow ones. + +// fastLink is the rate from which a device counts as close, and slowerBy how +// much slower the next device must be for this one to be waited for first. +const ( + fastLink = 4 << 20 + slowerBy = 4.0 + maxWaitFor = 20 * time.Minute +) + +// NotePeerPulled is told that a peer finished (or gave up) pulling a game from +// this device, and lets a sync waiting for it go on. +func (e *Engine) NotePeerPulled(gameID, peerID string) { + e.linkMu.Lock() + ch := e.pullWaits[gameID+"|"+peerID] + delete(e.pullWaits, gameID+"|"+peerID) + e.linkMu.Unlock() + if ch != nil { + close(ch) + } +} + +// waitForPull waits, after asking a close peer to take this device's save, +// until it has — before the slow devices are asked, so the save reaches the +// close one at full speed and the slow ones can then take it from whichever +// device is closest to them. Bounded by how long the transfer should take, +// and given up when the context ends. +func (e *Engine) waitForPull(ctx context.Context, gameID string, peer Peer, bytes int64) { + ch := make(chan struct{}) + e.linkMu.Lock() + if e.pullWaits == nil { + e.pullWaits = map[string]chan struct{}{} + } + e.pullWaits[gameID+"|"+peer.ID] = ch + e.linkMu.Unlock() + defer func() { + e.linkMu.Lock() + if e.pullWaits[gameID+"|"+peer.ID] == ch { + delete(e.pullWaits, gameID+"|"+peer.ID) + } + e.linkMu.Unlock() + }() + wait := 30*time.Second + time.Duration(3*float64(bytes)/e.expectedRate(peer)*float64(time.Second)) + if wait > maxWaitFor { + wait = maxWaitFor + } + select { + case <-ch: + case <-time.After(wait): + e.Log("info", fmt.Sprintf("%s has not finished taking %s yet; going on with the other devices", peer.Name, gameID)) + case <-ctx.Done(): + } +} + +// waitsForClose reports whether, having asked peers[i] to take this device's +// save, the sync should wait for it before going on: it is a close device, +// and a much slower one is still to come. +func (e *Engine) waitsForClose(peers []Peer, i int) bool { + r := e.expectedRate(peers[i]) + if r < fastLink { + return false + } + for _, p := range peers[i+1:] { + if e.expectedRate(p)*slowerBy <= r { + return true + } + } + return false +} diff --git a/internal/p2p/syncengine/linkspeed_test.go b/internal/p2p/syncengine/linkspeed_test.go new file mode 100644 index 00000000..3e8544dc --- /dev/null +++ b/internal/p2p/syncengine/linkspeed_test.go @@ -0,0 +1,106 @@ +package syncengine + +import ( + "context" + "testing" + "time" +) + +func TestLinkKind(t *testing.T) { + cases := map[string]string{ + "192.168.178.20": "lan", + "10.0.0.5": "lan", + "127.0.0.1": "lan", + "100.99.121.82": "vpn", + "203.0.113.9": "internet", + } + for addr, want := range cases { + if got := linkKind(Peer{Address: addr}); got != want { + t.Errorf("%s: %s, want %s", addr, got, want) + } + } + if got := linkKind(Peer{Address: "relay"}); got != "relay" { + t.Errorf("relay: %s", got) + } +} + +// Before anything is measured the kind of address orders the devices; once a +// link is measured, the measurement does. +func TestOrderBySpeed(t *testing.T) { + env := setupEngine(t) + e := env.engine + home := Peer{ID: "home", Address: "192.168.178.20"} + vpn := Peer{ID: "vpn", Address: "100.99.121.82"} + relay := Peer{ID: "relay", Address: "relay"} + + got := e.OrderBySpeed([]Peer{relay, vpn, home}) + if got[0].ID != "home" || got[1].ID != "vpn" || got[2].ID != "relay" { + t.Errorf("by address: %v", ids(got)) + } + + // A VPN link that turns out fast, and a home one that turns out slow. + e.noteLinkRate(vpn, 200<<20, 2*time.Second) + e.noteLinkRate(home, 1<<20, 10*time.Second) + got = e.OrderBySpeed([]Peer{home, vpn, relay}) + if got[0].ID != "vpn" { + t.Errorf("by measurement: %v", ids(got)) + } + // Remembered across restarts. + e2 := New(env.store, nil, env.transport) + if e2.Link(vpn).BytesPerSec == 0 { + t.Error("the measured link was not kept") + } +} + +func TestNoteLinkRate_IgnoresTinyTransfers(t *testing.T) { + env := setupEngine(t) + p := Peer{ID: "p", Address: "192.168.1.2"} + env.engine.noteLinkRate(p, 10<<10, 2*time.Second) + if env.engine.Link(p).BytesPerSec != 0 { + t.Error("a transfer of a few KB was taken as the link's speed") + } +} + +// A close device is waited for before a much slower one, and the wait ends as +// soon as it reports that it has finished. +func TestWaitForClose(t *testing.T) { + env := setupEngine(t) + e := env.engine + home := Peer{ID: "home", Address: "192.168.178.20"} + vpn := Peer{ID: "vpn", Address: "100.99.121.82"} + peers := e.OrderBySpeed([]Peer{vpn, home}) + if !e.waitsForClose(peers, 0) { + t.Fatal("a close device is not waited for before a slow one") + } + if e.waitsForClose(peers, 1) { + t.Error("the last device is waited for, with nothing after it") + } + if e.waitsForClose(e.OrderBySpeed([]Peer{home}), 0) { + t.Error("waited with no slower device to follow") + } + + done := make(chan struct{}) + start := time.Now() + go func() { + e.waitForPull(context.Background(), "game1", home, 1<<30) + close(done) + }() + time.Sleep(100 * time.Millisecond) + e.NotePeerPulled("game1", home.ID) + select { + case <-done: + if time.Since(start) > 5*time.Second { + t.Error("the wait did not end when the device reported") + } + case <-time.After(10 * time.Second): + t.Fatal("the wait did not end when the device reported") + } +} + +func ids(ps []Peer) []string { + out := make([]string, len(ps)) + for i, p := range ps { + out[i] = p.ID + } + return out +} diff --git a/internal/p2p/transport_lan.go b/internal/p2p/transport_lan.go index 7fae41da..54fa1edc 100644 --- a/internal/p2p/transport_lan.go +++ b/internal/p2p/transport_lan.go @@ -97,15 +97,59 @@ func (t *lanTransport) FetchBlocks(ctx context.Context, peer syncengine.Peer, re // No encodings advertised: on a LAN the wire is typically faster than the // compressor, so the bytes saved cost more than they're worth. Responses // are still decoded, so a peer that compresses anyway is handled. - err := t.postJSON(ctx, peer, peerURL(peer, "/blocks/"+ref.GameID), map[string]any{ + // + // Given time for what it asks for. "LAN" here is any direct address, + // a VPN one (Tailscale) included, and a few MB of blocks at a VPN relay's + // hundred-odd KB a second outlasted the fixed 30 seconds every request + // used to get: the request failed, and with it the whole pull, over and + // over. Sized like the relay's (slowLinkBytesPerSec), within the bulk + // client's ceiling. + expected := int64(len(blockIndices)) * int64(max(blockSize, 1)) + ctx, cancel := context.WithTimeout(ctx, 30*time.Second+time.Duration(expected/(slowLinkBytesPerSec/2))*time.Second) + defer cancel() + raw, err := json.Marshal(map[string]any{ "relPath": ref.RelPath, "root": ref.Root, "blockIndices": blockIndices, "blockSize": blockSize, - }, &resp) + }) + if err != nil { + return nil, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, peerURL(peer, "/blocks/"+ref.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 + } return decodeBlocks(resp.Blocks) } +// ProbeSpeed times fetching n bytes from the peer's speed test +// (syncengine.SpeedProber). The clock starts once the answer begins to +// arrive, so the figure is the link's rate rather than its round trip. +func (t *lanTransport) ProbeSpeed(ctx context.Context, peer syncengine.Peer, n int) (int64, time.Duration, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, peerURL(peer, fmt.Sprintf("/speedtest?n=%d", n)), nil) + if err != nil { + return 0, 0, err + } + t.sign(req, peer, nil) + resp, err := lanBulkClient.Do(req) + if err != nil { + return 0, 0, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return 0, 0, fmt.Errorf("peer returned %d", resp.StatusCode) + } + start := time.Now() + got, err := io.Copy(io.Discard, resp.Body) + if err != nil { + return got, time.Since(start), err + } + return got, time.Since(start), nil +} + // 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) { diff --git a/internal/p2p/transport_wan.go b/internal/p2p/transport_wan.go index 543f5f79..beb26478 100644 --- a/internal/p2p/transport_wan.go +++ b/internal/p2p/transport_wan.go @@ -3,6 +3,7 @@ package p2p import ( "context" "encoding/json" + "errors" "fmt" "time" @@ -169,6 +170,16 @@ func (t *routingTransport) FetchFileBatch(ctx context.Context, peer syncengine.P return b.FetchFileBatch(ctx, peer, gameID, root, files) } +// ProbeSpeed tests LAN links only (syncengine.SpeedProber); the relay is the +// slowest link whatever a test would say. +func (t *routingTransport) ProbeSpeed(ctx context.Context, peer syncengine.Peer, n int) (int64, time.Duration, error) { + p, ok := t.lan.(syncengine.SpeedProber) + if peer.Wan() || !ok { + return 0, 0, errors.New("no speed test over the relay") + } + return p.ProbeSpeed(ctx, peer, n) +} + func (t *routingTransport) DeleteRemote(ctx context.Context, peer syncengine.Peer, ref syncengine.FileRef) error { return t.pick(peer).DeleteRemote(ctx, peer, ref) } diff --git a/internal/p2p/wanclient_handlers.go b/internal/p2p/wanclient_handlers.go index 9a899c0c..d7525270 100644 --- a/internal/p2p/wanclient_handlers.go +++ b/internal/p2p/wanclient_handlers.go @@ -172,6 +172,7 @@ func (w *WanClient) handleMessage(ctx context.Context, msg RelayMessage) { if w.engine.Sync.Progress.OnSyncError != nil { w.engine.Sync.Progress.OnSyncError(msg.GameID, ev) } + w.engine.Sync.NotePeerPulled(msg.GameID, msg.From) } case "request": diff --git a/internal/store/migrations/0038_peer_links.sql b/internal/store/migrations/0038_peer_links.sql new file mode 100644 index 00000000..4cff6cfb --- /dev/null +++ b/internal/store/migrations/0038_peer_links.sql @@ -0,0 +1,15 @@ +-- How fast this device has measured its connection to each paired device: +-- from transfers it made, and from a short speed test when there is nothing +-- recent (see internal/p2p/syncengine/linkspeed.go). Syncs go to the fastest +-- connections first, so a save reaches the devices close to it before it +-- crawls over a slow relay — and those can then hand it on. +-- +-- bytes_per_sec is a running average; measured_ms when it last changed; +-- probed_ms when a speed test last ran, so a slow link is not tested over and +-- over. +CREATE TABLE peer_links ( + peer_id TEXT PRIMARY KEY, + bytes_per_sec REAL NOT NULL DEFAULT 0, + measured_ms INTEGER NOT NULL DEFAULT 0, + probed_ms INTEGER NOT NULL DEFAULT 0 +); diff --git a/internal/store/peer_links.go b/internal/store/peer_links.go new file mode 100644 index 00000000..d65b3a37 --- /dev/null +++ b/internal/store/peer_links.go @@ -0,0 +1,37 @@ +package store + +import "fmt" + +// PeerLink is how fast this device has measured its connection to a peer. +// See migrations/0038_peer_links.sql. +type PeerLink struct { + PeerID string `db:"peer_id"` + BytesPerSec float64 `db:"bytes_per_sec"` + MeasuredMs int64 `db:"measured_ms"` + ProbedMs int64 `db:"probed_ms"` +} + +// PeerLinks returns every recorded link, by peer. +func (s *Store) PeerLinks() (map[string]PeerLink, error) { + var rows []PeerLink + if err := s.db.Select(&rows, `SELECT * FROM peer_links`); err != nil { + return nil, fmt.Errorf("list peer links: %w", err) + } + out := make(map[string]PeerLink, len(rows)) + for _, r := range rows { + out[r.PeerID] = r + } + return out, nil +} + +// SavePeerLink records a peer's link, replacing what was there. +func (s *Store) SavePeerLink(l PeerLink) error { + _, err := s.db.Exec(`INSERT INTO peer_links (peer_id, bytes_per_sec, measured_ms, probed_ms) VALUES (?, ?, ?, ?) + ON CONFLICT(peer_id) DO UPDATE SET bytes_per_sec = excluded.bytes_per_sec, + measured_ms = excluded.measured_ms, probed_ms = excluded.probed_ms`, + l.PeerID, l.BytesPerSec, l.MeasuredMs, l.ProbedMs) + if err != nil { + return fmt.Errorf("save peer link %s: %w", l.PeerID, err) + } + return nil +} From d3f90dbea9219f386fd6776cc65cce1e89a5f8bb Mon Sep 17 00:00:00 2001 From: ChakraFusion <112588066+ChakraFusion@users.noreply.github.com> Date: Fri, 2 Oct 2026 13:07:29 +0200 Subject: [PATCH 06/10] sync: an unmeasured link is speed-tested again soon A speed test that failed - a peer not yet on a build that answers one - counted as done for six hours, so every link stayed unmeasured and the order fell back to address guesses all that time. A link never measured is now tested again ten minutes after a failed test. Co-Authored-By: Claude Opus 5.5 --- internal/p2p/syncengine/linkspeed.go | 14 ++++++- internal/p2p/syncengine/linkspeed_test.go | 47 +++++++++++++++++++++++ 2 files changed, 60 insertions(+), 1 deletion(-) diff --git a/internal/p2p/syncengine/linkspeed.go b/internal/p2p/syncengine/linkspeed.go index b19543b1..179e2eb9 100644 --- a/internal/p2p/syncengine/linkspeed.go +++ b/internal/p2p/syncengine/linkspeed.go @@ -44,6 +44,8 @@ const ( // measured it since; probeBytes how much a test moves. probeEvery = 6 * time.Hour probeBytes = 2 << 20 + // probeRetry is how soon a speed test that failed is tried again. + probeRetry = 10 * time.Minute ) // prior rates, by address kind, used until a link is measured. @@ -177,7 +179,13 @@ func (e *Engine) ProbeLinkIfDue(ctx context.Context, peer Peer) bool { e.loadLinksLocked() l := e.links[peer.ID] now := time.Now() - due := now.Sub(time.UnixMilli(l.MeasuredMs)) > probeEvery && now.Sub(time.UnixMilli(l.ProbedMs)) > probeEvery + // A link never measured is tried again soon after a test that failed; + // one measured before only when that has gone stale. + retryAfter := probeEvery + if l.BytesPerSec == 0 { + retryAfter = probeRetry + } + due := now.Sub(time.UnixMilli(l.MeasuredMs)) > probeEvery && now.Sub(time.UnixMilli(l.ProbedMs)) > retryAfter if due { l.PeerID = peer.ID l.ProbedMs = now.UnixMilli() @@ -194,6 +202,10 @@ func (e *Engine) ProbeLinkIfDue(ctx context.Context, peer Peer) bool { defer cancel() n, d, err := prober.ProbeSpeed(ctx, peer, probeBytes) if err != nil { + // A link never measured is tried again after probeRetry (see due + // above): the peer may simply not have been ready — or not yet on a + // build that answers a speed test, which once left every link + // unmeasured for hours. e.Log("info", fmt.Sprintf("speed test with %s did not finish: %v", peer.Name, err)) return true } diff --git a/internal/p2p/syncengine/linkspeed_test.go b/internal/p2p/syncengine/linkspeed_test.go index 3e8544dc..c87657ab 100644 --- a/internal/p2p/syncengine/linkspeed_test.go +++ b/internal/p2p/syncengine/linkspeed_test.go @@ -2,6 +2,7 @@ package syncengine import ( "context" + "errors" "testing" "time" ) @@ -97,6 +98,52 @@ func TestWaitForClose(t *testing.T) { } } +type probeTransport struct { + *fakeTransport + fail bool + calls int +} + +func (p *probeTransport) ProbeSpeed(ctx context.Context, peer Peer, n int) (int64, time.Duration, error) { + p.calls++ + if p.fail { + return 0, 0, errors.New("peer returned 404") + } + return int64(n), time.Second, nil +} + +// A speed test that fails — a peer not yet updated to answer one — is tried +// again after minutes, not hours; a measured link is not tested again soon. +func TestProbeLink_RetriesAFailureSoon(t *testing.T) { + env := setupEngine(t) + pt := &probeTransport{fakeTransport: env.transport, fail: true} + env.engine.Transport = pt + p := Peer{ID: "p", Name: "P", Address: "192.168.1.9"} + + if !env.engine.ProbeLinkIfDue(context.Background(), p) { + t.Fatal("an unmeasured link was not tested") + } + if env.engine.ProbeLinkIfDue(context.Background(), p) { + t.Fatal("tested again straight after a failure") + } + // Eleven minutes later. + env.engine.linkMu.Lock() + l := env.engine.links[p.ID] + l.ProbedMs = time.Now().Add(-11 * time.Minute).UnixMilli() + env.engine.links[p.ID] = l + env.engine.linkMu.Unlock() + pt.fail = false + if !env.engine.ProbeLinkIfDue(context.Background(), p) { + t.Fatal("a failed test was not tried again after ten minutes") + } + if env.engine.Link(p).BytesPerSec == 0 { + t.Fatal("the successful test was not recorded") + } + if env.engine.ProbeLinkIfDue(context.Background(), p) { + t.Error("a freshly measured link was tested again") + } +} + func ids(ps []Peer) []string { out := make([]string, len(ps)) for i, p := range ps { From e6d080724c3c91f6e0c16544e707c9a5c53bd4a4 Mon Sep 17 00:00:00 2001 From: ChakraFusion <112588066+ChakraFusion@users.noreply.github.com> Date: Fri, 2 Oct 2026 13:31:09 +0200 Subject: [PATCH 07/10] sync: a long pull measures its link every 30 seconds, not only at its end A pull of a large save over a slow link takes an hour, and the link's speed was recorded once, when it ended: all that time the device list said "not measured" and the sync order fell back to address guesses. The link is now measured over each 30-second window while the pull runs, and the rest when it ends. Co-Authored-By: Claude Opus 5.5 --- internal/p2p/syncengine/engine.go | 24 ++++++++++++++++++------ 1 file changed, 18 insertions(+), 6 deletions(-) diff --git a/internal/p2p/syncengine/engine.go b/internal/p2p/syncengine/engine.go index 19040386..1537dfde 100644 --- a/internal/p2p/syncengine/engine.go +++ b/internal/p2p/syncengine/engine.go @@ -1733,13 +1733,24 @@ func (e *Engine) pullFiles(ctx context.Context, peer Peer, gameID string, game s tracker := newProgressTracker(totalBytes) throttle := e.throttleFor(peer.Wan()) - // What this transfer says about the connection (linkspeed.go), measured - // however it ends: a pull cut off half-way still moved what it moved. - pullStarted := time.Now() - defer func() { + // What this transfer says about the connection (linkspeed.go): every + // measureWindow while it runs — a pull of a large save takes an hour over a + // slow link, and one measurement at its end left the link unmeasured all + // that time — and what is left when it ends, however it ends: a pull cut + // off half-way still moved what it moved. + const measureWindow = 30 * time.Second + var measureMu sync.Mutex + measuredAt, measuredBytes := time.Now(), int64(0) + measure := func(final bool) { moved, _, _ := tracker.stats() - e.noteLinkRate(peer, moved, time.Since(pullStarted)) - }() + measureMu.Lock() + defer measureMu.Unlock() + if d := time.Since(measuredAt); final || d >= measureWindow { + e.noteLinkRate(peer, moved-measuredBytes, d) + measuredAt, measuredBytes = time.Now(), moved + } + } + defer measure(true) // Progress reporter shared by the per-file loop and the block-group // loop inside each file. Without in-file reporting, a single large @@ -1756,6 +1767,7 @@ func (e *Engine) pullFiles(ctx context.Context, peer Peer, gameID string, game s } lastReport = time.Now() reportMu.Unlock() + measure(false) bytesPulled, speed, pct := tracker.stats() ev := ProgressEvent{PeerName: peer.Name, BytesTransferred: bytesPulled, TotalBytes: totalBytes, SpeedBytesPerSec: speed, Percentage: pct} From 9e286154be62af8652506b7a3ae28e752654d0e1 Mon Sep 17 00:00:00 2001 From: ChakraFusion <112588066+ChakraFusion@users.noreply.github.com> Date: Fri, 2 Oct 2026 14:50:43 +0200 Subject: [PATCH 08/10] sync: a fast link's speed test is kept, not discarded as too short Over a home network the 2 MB speed test is over in a fraction of a second, and the rule that ignores transfers under half a second - meant for syncs, whose small transfers say more about round trips than about the link - threw it away: the fastest link stayed unmeasured, and with nothing measured it sorted behind the measured slow ones. A test that quick is now run again at 8 MB, and a test counts however short it is. Co-Authored-By: Claude Opus 5.5 --- internal/p2p/syncengine/linkspeed.go | 16 ++++++++++++++-- internal/p2p/syncengine/linkspeed_test.go | 22 ++++++++++++++++++++++ 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/internal/p2p/syncengine/linkspeed.go b/internal/p2p/syncengine/linkspeed.go index 179e2eb9..21af61f7 100644 --- a/internal/p2p/syncengine/linkspeed.go +++ b/internal/p2p/syncengine/linkspeed.go @@ -125,7 +125,12 @@ func (e *Engine) expectedRate(peer Peer) float64 { // noteLinkRate records a transfer with a peer: bytes in d. func (e *Engine) noteLinkRate(peer Peer, bytes int64, d time.Duration) { - if bytes < minMeasureBytes || d < minMeasureTime { + e.recordLinkRate(peer, bytes, d, minMeasureTime) +} + +// recordLinkRate records bytes in d, when d is at least minTime. +func (e *Engine) recordLinkRate(peer Peer, bytes int64, d, minTime time.Duration) { + if bytes < minMeasureBytes || d < minTime || d <= 0 { return } rate := float64(bytes) / d.Seconds() @@ -201,6 +206,11 @@ func (e *Engine) ProbeLinkIfDue(ctx context.Context, peer Peer) bool { ctx, cancel := context.WithTimeout(ctx, 60*time.Second) defer cancel() n, d, err := prober.ProbeSpeed(ctx, peer, probeBytes) + if err == nil && d < minMeasureTime { + // Over a home network the test is over before it measures much: + // once more, larger, for a figure worth having. + n, d, err = prober.ProbeSpeed(ctx, peer, 4*probeBytes) + } if err != nil { // A link never measured is tried again after probeRetry (see due // above): the peer may simply not have been ready — or not yet on a @@ -209,7 +219,9 @@ func (e *Engine) ProbeLinkIfDue(ctx context.Context, peer Peer) bool { e.Log("info", fmt.Sprintf("speed test with %s did not finish: %v", peer.Name, err)) return true } - e.noteLinkRate(peer, n, d) + // A test is a measurement however short: it moved nothing but the test, + // so no round trips of a sync's own are in it. + e.recordLinkRate(peer, n, d, time.Millisecond) e.Log("info", fmt.Sprintf("connection to %s: %s/s (%s)", peer.Name, humanBytes(int64(e.Link(peer).BytesPerSec)), linkKind(peer))) return true } diff --git a/internal/p2p/syncengine/linkspeed_test.go b/internal/p2p/syncengine/linkspeed_test.go index c87657ab..968dcc87 100644 --- a/internal/p2p/syncengine/linkspeed_test.go +++ b/internal/p2p/syncengine/linkspeed_test.go @@ -102,6 +102,8 @@ type probeTransport struct { *fakeTransport fail bool calls int + // perMB is how long each MB takes; a second when unset. + perMB time.Duration } func (p *probeTransport) ProbeSpeed(ctx context.Context, peer Peer, n int) (int64, time.Duration, error) { @@ -109,6 +111,9 @@ func (p *probeTransport) ProbeSpeed(ctx context.Context, peer Peer, n int) (int6 if p.fail { return 0, 0, errors.New("peer returned 404") } + if p.perMB > 0 { + return int64(n), time.Duration(n>>20) * p.perMB, nil + } return int64(n), time.Second, nil } @@ -144,6 +149,23 @@ func TestProbeLink_RetriesAFailureSoon(t *testing.T) { } } +// A home network finishes the test before it has measured much: it is run +// again larger, and the result is kept — not discarded as too short, which +// left the fastest link unmeasured and sorted behind the slow ones. +func TestProbeLink_AFastLinkIsMeasured(t *testing.T) { + env := setupEngine(t) + pt := &probeTransport{fakeTransport: env.transport, perMB: 20 * time.Millisecond} + env.engine.Transport = pt + p := Peer{ID: "lan", Name: "LAN", Address: "192.168.178.20"} + env.engine.ProbeLinkIfDue(context.Background(), p) + if got := env.engine.Link(p).BytesPerSec; got < 20<<20 { + t.Errorf("a 50 MB/s link measured as %.0f B/s", got) + } + if pt.calls != 2 { + t.Errorf("%d test(s), want the short one and a larger one", pt.calls) + } +} + func ids(ps []Peer) []string { out := make([]string, len(ps)) for i, p := range ps { From ecb16c9a6041afe70e0588d6995422fcd655502f Mon Sep 17 00:00:00 2001 From: ChakraFusion <112588066+ChakraFusion@users.noreply.github.com> Date: Thu, 15 Oct 2026 14:51:50 +0200 Subject: [PATCH 09/10] store: number the peer links migration 0039 0038 is taken by 0038_peer_folder_trust on dev, which reaches dev-public later; the maintainer asked for 0039 here (#21). Co-Authored-By: Claude Opus 5.5 --- .../migrations/{0038_peer_links.sql => 0039_peer_links.sql} | 0 internal/store/peer_links.go | 2 +- 2 files changed, 1 insertion(+), 1 deletion(-) rename internal/store/migrations/{0038_peer_links.sql => 0039_peer_links.sql} (100%) diff --git a/internal/store/migrations/0038_peer_links.sql b/internal/store/migrations/0039_peer_links.sql similarity index 100% rename from internal/store/migrations/0038_peer_links.sql rename to internal/store/migrations/0039_peer_links.sql diff --git a/internal/store/peer_links.go b/internal/store/peer_links.go index d65b3a37..ddff1ddd 100644 --- a/internal/store/peer_links.go +++ b/internal/store/peer_links.go @@ -3,7 +3,7 @@ package store import "fmt" // PeerLink is how fast this device has measured its connection to a peer. -// See migrations/0038_peer_links.sql. +// See migrations/0039_peer_links.sql. type PeerLink struct { PeerID string `db:"peer_id"` BytesPerSec float64 `db:"bytes_per_sec"` From 93ce21d8d7626a917474e99897bf32918c22fe52 Mon Sep 17 00:00:00 2001 From: ChakraFusion <112588066+ChakraFusion@users.noreply.github.com> Date: Tue, 20 Oct 2026 21:48:37 +0200 Subject: [PATCH 10/10] i18n: the device link line and its tooltip are said from the catalog (en, zh-Hans, zh-Hant) How a device is reached, its measured speed and the tooltip on why that matters were English strings in Devices.svelte. They are devices.link* keys now, and linkLabel takes $t so the line follows a change of language. As asked on #73. Co-Authored-By: Claude Opus 5.5 --- cmd/opensave-app/frontend/src/locales/en.json | 9 ++++++++- .../frontend/src/locales/zh-Hans.json | 9 ++++++++- .../frontend/src/locales/zh-Hant.json | 9 ++++++++- .../frontend/src/views/Devices.svelte | 16 +++++++++------- 4 files changed, 33 insertions(+), 10 deletions(-) diff --git a/cmd/opensave-app/frontend/src/locales/en.json b/cmd/opensave-app/frontend/src/locales/en.json index 37e34a1f..cfd89421 100644 --- a/cmd/opensave-app/frontend/src/locales/en.json +++ b/cmd/opensave-app/frontend/src/locales/en.json @@ -258,5 +258,12 @@ "transfers.empty": "Nothing has moved between your devices since OpenSave started.", "transfers.now": "Now", "transfers.recently": "Recently", - "transfers.readError": "Couldn't read transfers: {error}" + "transfers.readError": "Couldn't read transfers: {error}", + "devices.linkLan": "home network", + "devices.linkVpn": "VPN (e.g. Tailscale)", + "devices.linkRelay": "internet relay", + "devices.linkInternet": "internet", + "devices.linkUnmeasured": "{kind} · speed not measured yet", + "devices.linkMeasured": "{kind} · {rate} (measured {when})", + "devices.linkTitle": "Saves go to the fastest connections first, and a device that is behind takes the newer save from the fastest device that has it." } diff --git a/cmd/opensave-app/frontend/src/locales/zh-Hans.json b/cmd/opensave-app/frontend/src/locales/zh-Hans.json index 61697d1a..9c171264 100644 --- a/cmd/opensave-app/frontend/src/locales/zh-Hans.json +++ b/cmd/opensave-app/frontend/src/locales/zh-Hans.json @@ -225,5 +225,12 @@ "transfers.empty": "自 OpenSave 启动以来，设备之间还没有传输过任何内容。", "transfers.now": "进行中", "transfers.recently": "最近", - "transfers.readError": "无法读取传输记录：{error}" + "transfers.readError": "无法读取传输记录：{error}", + "devices.linkLan": "家庭网络", + "devices.linkVpn": "VPN（如 Tailscale）", + "devices.linkRelay": "互联网中继", + "devices.linkInternet": "互联网", + "devices.linkUnmeasured": "{kind} · 尚未测速", + "devices.linkMeasured": "{kind} · {rate}（测于 {when}）", + "devices.linkTitle": "存档会优先通过最快的连接发送；落后的设备会从拥有较新存档的设备中最快的那台获取。" } diff --git a/cmd/opensave-app/frontend/src/locales/zh-Hant.json b/cmd/opensave-app/frontend/src/locales/zh-Hant.json index 4db04909..ad72b4bb 100644 --- a/cmd/opensave-app/frontend/src/locales/zh-Hant.json +++ b/cmd/opensave-app/frontend/src/locales/zh-Hant.json @@ -225,5 +225,12 @@ "transfers.empty": "自 OpenSave 啟動以來，裝置之間還沒有傳輸過任何內容。", "transfers.now": "進行中", "transfers.recently": "最近", - "transfers.readError": "無法讀取傳輸記錄：{error}" + "transfers.readError": "無法讀取傳輸記錄：{error}", + "devices.linkLan": "家用網路", + "devices.linkVpn": "VPN（如 Tailscale）", + "devices.linkRelay": "網際網路中繼", + "devices.linkInternet": "網際網路", + "devices.linkUnmeasured": "{kind} · 尚未測速", + "devices.linkMeasured": "{kind} · {rate}（測於 {when}）", + "devices.linkTitle": "存檔會優先透過最快的連線傳送；落後的裝置會從擁有較新存檔的裝置中最快的那臺取得。" } diff --git a/cmd/opensave-app/frontend/src/views/Devices.svelte b/cmd/opensave-app/frontend/src/views/Devices.svelte index 4b53ad09..34884005 100644 --- a/cmd/opensave-app/frontend/src/views/Devices.svelte +++ b/cmd/opensave-app/frontend/src/views/Devices.svelte @@ -9,6 +9,7 @@ import Globe from 'lucide-svelte/icons/globe'; import Network from 'lucide-svelte/icons/network'; import Download from 'lucide-svelte/icons/download'; + import { t } from '../lib/i18n.js'; export let params = {}; @@ -78,12 +79,13 @@ const fmtTime = (t) => (t ? new Date(t).toLocaleString() : 'never'); // How this device reaches the peer, and how fast it has found that to be. - const linkKinds = { lan: 'home network', vpn: 'VPN (e.g. Tailscale)', relay: 'internet relay', internet: 'internet' }; + // linkLabel is given $t, so the line follows a change of language. + const linkKinds = { lan: 'devices.linkLan', vpn: 'devices.linkVpn', relay: 'devices.linkRelay', internet: 'devices.linkInternet' }; const fmtRate = (b) => (b >= 1048576 ? (b / 1048576).toFixed(1) + ' MB/s' : Math.max(1, Math.round(b / 1024)) + ' KB/s'); - function linkLabel(link) { - const kind = linkKinds[link.kind] ?? link.kind; - if (!link.bytesPerSec) return `${kind} · speed not measured yet`; - return `${kind} · ${fmtRate(link.bytesPerSec)} (measured ${fmtTime(link.measuredMs)})`; + function linkLabel(link, tr) { + const kind = linkKinds[link.kind] ? tr(linkKinds[link.kind]) : link.kind; + if (!link.bytesPerSec) return tr('devices.linkUnmeasured', { kind }); + return tr('devices.linkMeasured', { kind, rate: fmtRate(link.bytesPerSec), when: fmtTime(link.measuredMs) }); } @@ -115,8 +117,8 @@ {#if peer.appVersion}· OpenSave {peer.appVersion}{/if}
{#if peer.link} -
- {linkLabel(peer.link)} +
+ {linkLabel(peer.link, $t)}
{/if}