sync: a peer that pulled says what it now holds, so the base is recorded at once

e2e/quick_turns_test.go

@@ -0,0 +1,44 @@
+package e2e
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/opensave/opensave/testutil"
+)
+
+// Two devices take turns quickly over the relay: each edits as soon as the
+// other's change has arrived, before the report that it arrived has been
+// acted on. Nobody edits at the same time, so no turn may be a conflict.
+//
+// The device that handed a change over records "we agree on this" only once
+// it hears the other now holds it. It used to hear only which files were
+// taken, and to check the rest by fetching the other's save afterwards — by
+// which time the other had already edited again. The agreement was never
+// recorded, and the next turn read as both sides having changed.
+func TestQuickTurns_NoFalseConflict(t *testing.T) {
+	a, b, gameID := trackBothOverRelay(t, "QuickTurns", map[string]string{"slot1.sav": "turn-0"})
+	logOnFailure(t, a, b)
+	testutil.SettleSync(t, gameID, a, b)
+
+	writer, other := b, a
+	for turn := 1; turn <= 6; turn++ {
+		want := fmt.Sprintf("turn-%d", turn)
+		writer.WriteSave("slot1.sav", want)
+		if status, _ := syncTo(writer, gameID, other.NodeID()); status == "conflict" {
+			c, _ := conflictOn(writer, gameID)
+			t.Fatalf("turn %d: a one-sided edit conflicted: %+v", turn, c.DiffFiles)
+		}
+		if !testutil.WaitFor(60*time.Second, func() bool { return other.ReadSave("slot1.sav") == want }) {
+			if c, ok := conflictOn(other, gameID); ok {
+				t.Fatalf("turn %d: the edit never arrived; the receiving device holds a conflict: %+v", turn, c.DiffFiles)
+			}
+			t.Fatalf("turn %d: the edit never arrived", turn)
+		}
+		if c, ok := conflictOn(writer, gameID); ok {
+			t.Fatalf("turn %d: the device that edited holds a conflict: %+v", turn, c.DiffFiles)
+		}
+		writer, other = other, writer // the other edits at once
+	}
+}

internal/p2p/routes.go

@@ -924,6 +924,12 @@ func (e *Engine) handleSyncEvent(w http.ResponseWriter, r *http.Request) {
 		if peer, ok := e.peerByAddress(clientIP(r)); ok {
 			e.peerFinishedPulling(gameID, peer, body.Data)
 		}
+	case "agreed":
+		// The peer pulled from us and now holds exactly this state.
+		if peer, ok := e.peerByAddress(clientIP(r)); ok {
+			hash, _ := body.Data["manifestHash"].(string)
+			e.Sync.NotePeerHolds(e.localGameID(gameID), peer, hash)
+		}
 	case "in-sync":
 		// The peer verified both sides hold identical content; confirm on
 		// our side (hash re-check) before recording lineage + last-synced.

internal/p2p/syncengine/agreed_report_test.go

@@ -0,0 +1,95 @@
+package syncengine
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/opensave/opensave/internal/delta"
+)
+
+// The state this device (B) ends up in when the peer (A) pulled B's edit and
+// edited again before B recorded that A had taken it: B's base is still the
+// state before its own edit, its push record names its edit, and A holds a
+// newer save. Nobody edited at the same time.
+func setupPeerEditedAgain(t *testing.T) (*engineEnv, string) {
+	t.Helper()
+	env := setupEngine(t)
+	write(t, env.localDir, "save.dat", "turn-1")
+	write(t, env.remoteDir, "save.dat", "turn-1")
+	base, err := delta.BuildManifest(env.localDir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := env.store.SetAgreedHash("game1", env.peer.ID, base.ManifestHash()); err != nil {
+		t.Fatal(err)
+	}
+
+	// B edits and hands it over.
+	write(t, env.localDir, "save.dat", "turn-2")
+	handed, err := delta.BuildManifest(env.localDir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := env.store.SetPushedHash("game1", env.peer.ID, handed.ManifestHash()); err != nil {
+		t.Fatal(err)
+	}
+
+	// A takes it, then edits again at once.
+	write(t, env.remoteDir, "save.dat", "turn-2")
+	later := time.Now().Add(2 * time.Second)
+	write(t, env.remoteDir, "save.dat", "turn-3")
+	_ = os.Chtimes(filepath.Join(env.remoteDir, "save.dat"), later, later)
+	return env, handed.ManifestHash()
+}
+
+// Without A saying what it took, B reads both sides as changed: the false
+// conflict this fixes, kept here so the test shows it is real.
+func TestAgreedReport_WithoutItTheNextSyncConflicts(t *testing.T) {
+	env, _ := setupPeerEditedAgain(t)
+	res, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Status != "conflict" {
+		t.Skipf("no conflict without the report (%s); nothing to compare against", res.Status)
+	}
+}
+
+// With A's report that it holds exactly what B handed over, B records that as
+// the base, and A's later edit is just a newer save to take.
+func TestAgreedReport_ThePeersNextEditIsTakenWithoutAConflict(t *testing.T) {
+	env, handed := setupPeerEditedAgain(t)
+	env.engine.NotePeerHolds("game1", env.peer, handed)
+	if got := env.store.GetAgreedHash("game1", env.peer.ID); got != handed {
+		t.Fatalf("base = %q after the report, want the state handed over", got)
+	}
+	res, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Status == "conflict" {
+		t.Fatal("a conflict over edits made one after the other")
+	}
+	got, err := os.ReadFile(filepath.Join(env.localDir, "save.dat"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(got) != "turn-3" {
+		t.Errorf("save = %q, want the peer's newer turn-3", got)
+	}
+}
+
+// A report of any other state moves nothing: the base only goes to a state
+// both sides are known to have held.
+func TestAgreedReport_AnotherStateIsIgnored(t *testing.T) {
+	env, handed := setupPeerEditedAgain(t)
+	before := env.store.GetAgreedHash("game1", env.peer.ID)
+	env.engine.NotePeerHolds("game1", env.peer, "not-"+handed)
+	env.engine.NotePeerHolds("game1", env.peer, "")
+	if got := env.store.GetAgreedHash("game1", env.peer.ID); got != before {
+		t.Errorf("base moved to %q on a report of a state never handed over", got)
+	}
+}

internal/p2p/syncengine/engine.go

@@ -827,6 +827,15 @@ func (e *Engine) SyncWithPeer(ctx context.Context, gameID string, peer Peer) (Re
 	case !decision.HasPush() &&
 		len(decision.FilesToDeleteOnPeer) == 0 && len(decision.DirsToDeleteOnPeer) == 0:
 		_ = e.Store.SetAgreedHash(gameID, peer.ID, remoteData.Manifest.ManifestHash())
+		// And the peer is told, so it can record the same agreement at once
+		// (NotePeerHolds) rather than only by fetching this side's save again
+		// later — by which time this side may have edited it, and the
+		// agreement was never recorded there: its next sync then read both
+		// sides as changed, a conflict over edits made one after the other.
+		// A peer that predates this ignores the report.
+		e.Transport.ReportSyncEvent(peer, gameID, "agreed", map[string]any{
+			"manifestHash": remoteData.Manifest.ManifestHash(),
+		})
 	case handedOver != "":
 		// What the peer was actually offered, from step 9.
 		_ = e.Store.SetPushedHash(gameID, peer.ID, handedOver)
@@ -2029,3 +2038,16 @@ func (e *Engine) recordedDeletions(gameID, root string) map[string]DeletedRecord
 	}
 	return out
 }
+
+// NotePeerHolds records a peer's report that, having pulled from this device,
+// it now holds exactly hash. When that is the state this device handed over
+// (the push record), both held it, so it is the agreed base — recorded now,
+// not only once a later look at the peer's save finds it unchanged, which a
+// peer that has already edited again never is. Anything else is ignored: the
+// base only ever moves to a state both sides are known to have held.
+func (e *Engine) NotePeerHolds(gameID string, peer Peer, hash string) {
+	if hash == "" || hash != e.Store.GetPushedHash(gameID, peer.ID) {
+		return
+	}
+	_ = e.Store.SetAgreedHash(gameID, peer.ID, hash)
+}

internal/p2p/syncengine/engine_test.go

@@ -218,9 +218,11 @@ func TestSync_PullNewRemoteFile(t *testing.T) {
 	if len(env.transport.syncEvents) == 0 || env.transport.syncEvents[0] != "sync-start" {
 		t.Errorf("expected sync-start event first, got %v", env.transport.syncEvents)
 	}
-	last := env.transport.syncEvents[len(env.transport.syncEvents)-1]
-	if last != "sync-complete" {
-		t.Errorf("expected sync-complete last, got %v", env.transport.syncEvents)
+	// sync-complete ends the progress; after a pure pull the peer is then told
+	// which state this side now holds (NotePeerHolds).
+	evs := env.transport.syncEvents
+	if len(evs) < 2 || evs[len(evs)-2] != "sync-complete" || evs[len(evs)-1] != "agreed" {
+		t.Errorf("expected sync-complete then agreed last, got %v", evs)
 	}
 }
 

internal/p2p/wanclient_handlers.go

@@ -152,6 +152,17 @@ func (w *WanClient) handleMessage(ctx context.Context, msg RelayMessage) {
 				_ = json.Unmarshal(msg.Data, &raw)
 				w.engine.peerFinishedPulling(msg.GameID, sp, raw)
 			}
+		case "agreed":
+			// The peer pulled from us over the relay and now holds exactly
+			// this state; handled as on the LAN.
+			var payload struct {
+				ManifestHash string `json:"manifestHash"`
+			}
+			_ = json.Unmarshal(msg.Data, &payload)
+			if peer, err := w.engine.Store.GetPeer(msg.From); err == nil {
+				sp := syncengine.Peer{ID: peer.ID, Name: peer.Name, Address: "relay", Port: peer.Port, IsWan: true}
+				w.engine.Sync.NotePeerHolds(w.engine.localGameID(msg.GameID), sp, payload.ManifestHash)
+			}
 		case "in-sync":
 			// Peer verified both sides match: confirm on our side (hash
 			// re-check) before recording lineage + last-synced.