sync: batch small files, skip unchanged manifests, probe presence on its own

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))
+		})
+	}
+}

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)
+	}
+}

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,12 @@ 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
+	// 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
@@ -283,6 +293,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)
@@ -379,9 +390,34 @@ 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) {
+	// 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
@@ -414,11 +450,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"
 		}
 
@@ -542,6 +585,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) {
@@ -571,6 +616,43 @@ 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) {
+		// 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 {
+			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) {
@@ -602,6 +684,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 {
@@ -616,6 +705,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() {
@@ -633,6 +738,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

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
+}