sync: two devices pushing to each other is a conflict, not a loop

e2e/reported_conflict_wan_test.go

@@ -1,6 +1,7 @@
 package e2e
 
 import (
+	"fmt"
 	"net/http"
 	"testing"
 	"time"
@@ -91,24 +92,33 @@ func TestReportedWAN_AlternatingEditsMustNotConflict(t *testing.T) {
 	// Several turns each way. One device edits, it propagates, the other
 	// edits. Nobody ever edits at the same time, so nothing here is a
 	// genuine divergence and no turn may raise a conflict.
+	//
+	// Each turn writes something new. It used to write the same "a-turn" and
+	// "b-turn" every round, which is a different case: bytes coming back that
+	// a device's base, a round behind, still names. Then neither side can tell
+	// from hashes which one moved, and the two devices disagree (#71). That
+	// case is now a conflict rather than an endless exchange of pushes, and is
+	// tested on its own (syncengine's TestPushLoop_*).
 	for round := 0; round < 3; round++ {
 		time.Sleep(syncSettleWindow)
-		a.WriteSave("slot1.sav", "a-turn")
+		aTurn := fmt.Sprintf("a-turn %d", round)
+		a.WriteSave("slot1.sav", aTurn)
 		if status, _ := syncTo(a, gameID, b.NodeID()); status == "conflict" {
 			ca, _ := conflictOn(a, gameID)
 			t.Fatalf("round %d: A's turn conflicted: %+v", round, ca.DiffFiles)
 		}
-		if !testutil.WaitFor(60*time.Second, func() bool { return b.ReadSave("slot1.sav") == "a-turn" }) {
+		if !testutil.WaitFor(60*time.Second, func() bool { return b.ReadSave("slot1.sav") == aTurn }) {
 			t.Fatalf("round %d: A's edit never reached B", round)
 		}
 
 		time.Sleep(syncSettleWindow)
-		b.WriteSave("slot1.sav", "b-turn")
+		bTurn := fmt.Sprintf("b-turn %d", round)
+		b.WriteSave("slot1.sav", bTurn)
 		if status, _ := syncTo(b, gameID, a.NodeID()); status == "conflict" {
 			cb, _ := conflictOn(b, gameID)
 			t.Fatalf("round %d: B's turn conflicted: %+v", round, cb.DiffFiles)
 		}
-		if !testutil.WaitFor(60*time.Second, func() bool { return a.ReadSave("slot1.sav") == "b-turn" }) {
+		if !testutil.WaitFor(60*time.Second, func() bool { return a.ReadSave("slot1.sav") == bTurn }) {
 			t.Fatalf("round %d: B's edit never reached A", round)
 		}
 	}

internal/p2p/routes.go

@@ -965,6 +965,7 @@ func (e *Engine) peerByAddress(ip string) (syncengine.Peer, bool) {
 // newer content for us to pull.
 func (e *Engine) handleSyncTrigger(w http.ResponseWriter, r *http.Request) {
 	gameID := chi.URLParam(r, "gameId")
+	e.Sync.NoteTriggered(gameID, lanPeerID(r)) // syncengine/pushloop.go
 	e.GoSync(func(ctx context.Context) {
 		ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
 		defer cancel()

internal/p2p/syncengine/engine.go

@@ -131,6 +131,12 @@ type Engine struct {
 	// each peer, per game (served.go).
 	servedMu sync.Mutex
 	served   map[string][]servedState
+	// loopMu guards the push-loop watch: per game and peer, when the peer last
+	// asked this device to pull, and the push this device last answered with
+	// (pushloop.go).
+	loopMu    sync.Mutex
+	triggered map[string]time.Time
+	pushBacks map[string]pushBack
 
 	mu           sync.Mutex
 	activeSyncs  map[string]bool
@@ -694,6 +700,17 @@ func (e *Engine) SyncWithPeer(ctx context.Context, gameID string, peer Peer) (Re
 	}
 	decision := ComputeWithDeletions(localManifest, remoteData.Manifest, lineageFiles, lineageDirs, agreedHash, deleted)
 
+	// The peer asked this device to pull, and this device would answer by
+	// asking it to pull again, with neither side changed since: the two
+	// disagree about which save is newer, and would go on asking each other
+	// forever (pushloop.go). Both versions stay; the user decides.
+	if e.pushLoop(gameID, peer.ID, decision, localManifest.ManifestHash(), remoteData.Manifest.ManifestHash()) {
+		e.Log("warn", fmt.Sprintf("%q and %q each hold %q as newer than the other's and keep asking each other to take it — holding it for a decision instead",
+			e.deviceName(), peer.Name, game.Name))
+		e.registerConflict(gameID, peer, localManifest, remoteData)
+		return Result{Status: "conflict", PeerID: peer.ID, PeerName: peer.Name}, nil
+	}
+
 	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)

internal/p2p/syncengine/pushloop.go

@@ -0,0 +1,102 @@
+package syncengine
+
+import (
+	"time"
+)
+
+// Two devices pushing to each other forever (#71).
+//
+// The agreed base decides which side moved (#42): a side whose whole manifest
+// is the base has not changed. That is right except when the same bytes come
+// back after the base was agreed. One device's base can still be a round
+// behind — the report of the other's pull not yet processed — and a save
+// written back to exactly that older state then reads as "unchanged" there.
+// That device pushes, the other correctly sees itself as newer and pushes too,
+// neither pulls, nothing moves either base, and they ask each other to pull
+// several times a second. The change never arrives.
+//
+// Hashes cannot tell an unchanged save from the same bytes written again, and
+// clocks cannot either (#42 exists because a restored copy of the base has a
+// newer mtime). So the rule stays and the loop is caught instead: a peer that
+// was just told to pull and answered by asking this device to pull, with
+// neither side's content changed since, has decided the opposite of this
+// device about the same two states. That is two devices disagreeing about
+// which save is newer, which is what a conflict is. It is raised as one: both
+// versions are kept and the user is asked, rather than either side guessing.
+//
+// One answered push is not yet a loop: a peer can decide to push while this
+// device's trigger is still on its way. pushLoopRounds of them, each answered
+// with the same two states, is.
+
+// pushBack is the push this device last decided for a game and peer: the two
+// states it was decided on, and how many times in a row the peer has answered
+// it by asking this device to pull, without either state changing.
+type pushBack struct {
+	local, remote string
+	rounds        int
+}
+
+// pushLoopRounds is how many answered pushes in a row make a loop. The loop
+// runs several times a second, so stopping at two costs nothing, and one
+// would mistake a peer's crossing trigger for a refusal.
+const pushLoopRounds = 2
+
+// pushLoopWindow bounds how long ago the peer's request to pull may have
+// arrived to count as the answer to this device's push.
+const pushLoopWindow = 2 * time.Minute
+
+func loopKey(gameID, peerID string) string { return gameID + "\x00" + peerID }
+
+// NoteTriggered records that peerID asked this device to pull gameID.
+func (e *Engine) NoteTriggered(gameID, peerID string) {
+	if peerID == "" {
+		return
+	}
+	e.loopMu.Lock()
+	defer e.loopMu.Unlock()
+	if e.triggered == nil {
+		e.triggered = map[string]time.Time{}
+	}
+	e.triggered[loopKey(gameID, peerID)] = time.Now()
+}
+
+// pushLoop is told the decision of a sync with a peer, and reports whether the
+// two devices are pushing to each other without either changing. Anything but
+// a push — a pull, a deletion, agreement — ends the watch, and so does either
+// side's content changing.
+func (e *Engine) pushLoop(gameID, peerID string, d Decision, localHash, remoteHash string) bool {
+	e.loopMu.Lock()
+	defer e.loopMu.Unlock()
+	key := loopKey(gameID, peerID)
+	// The peer's request is used up by this sync, whatever it decides: the
+	// next one counts only if the peer asks again.
+	at, asked := e.triggered[key]
+	delete(e.triggered, key)
+	asked = asked && time.Since(at) < pushLoopWindow
+
+	pushOnly := d.HasPush() && !d.HasPull() && !d.HasDeletions()
+	if !pushOnly {
+		delete(e.pushBacks, key)
+		return false
+	}
+	if e.pushBacks == nil {
+		e.pushBacks = map[string]pushBack{}
+	}
+	prev, seen := e.pushBacks[key]
+	if !seen || prev.local != localHash || prev.remote != remoteHash {
+		e.pushBacks[key] = pushBack{local: localHash, remote: remoteHash}
+		return false
+	}
+	if !asked {
+		// The same push again, but not as the peer's answer: a retry of this
+		// device's own, after a pull on the peer that did not happen yet.
+		return false
+	}
+	prev.rounds++
+	e.pushBacks[key] = prev
+	if prev.rounds >= pushLoopRounds {
+		delete(e.pushBacks, key)
+		return true
+	}
+	return false
+}

internal/p2p/syncengine/pushloop_test.go

@@ -0,0 +1,127 @@
+package syncengine
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/opensave/opensave/internal/delta"
+)
+
+// aba sets up this device's side of #71's reproduction: the two devices once
+// held the same save, this device now holds "a-turn", the peer holds "b-turn",
+// and the agreed base is the peer's state — bytes the peer held before and
+// wrote again, which #42's rule reads as "the peer has not changed", so this
+// device decides to push.
+func aba(t *testing.T) *engineEnv {
+	t.Helper()
+	env := setupEngine(t)
+	write(t, env.localDir, "slot1.sav", "start")
+	write(t, env.remoteDir, "slot1.sav", "start")
+	if res, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer); err != nil || res.Status != "in_sync" {
+		t.Fatalf("setup: %+v %v", res, err)
+	}
+	write(t, env.localDir, "slot1.sav", "a-turn")
+	write(t, env.remoteDir, "slot1.sav", "b-turn")
+	remote, err := delta.BuildManifest(env.remoteDir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := env.store.SetAgreedHash("game1", env.peer.ID, remote.ManifestHash()); err != nil {
+		t.Fatal(err)
+	}
+	return env
+}
+
+// The peer answers every push by asking this device to pull, neither side
+// changing: the loop in #71. It used to go on forever; now the second answered
+// push is a conflict, with both saves as they were and no further trigger.
+func TestPushLoop_TwoDevicesPushingToEachOtherIsAConflict(t *testing.T) {
+	env := aba(t)
+	ctx := context.Background()
+
+	res, err := env.engine.SyncWithPeer(ctx, "game1", env.peer)
+	if err != nil || res.Status != "triggered_peer_pull" {
+		t.Fatalf("this device's own push: %+v %v", res, err)
+	}
+	for round := 1; round <= pushLoopRounds; round++ {
+		// The peer pushed back instead of pulling.
+		env.engine.NoteTriggered("game1", env.peer.ID)
+		res, err = env.engine.SyncWithPeer(ctx, "game1", env.peer)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if round < pushLoopRounds && res.Status != "triggered_peer_pull" {
+			t.Fatalf("answered push %d: %+v, want another push before it counts as a loop", round, res)
+		}
+	}
+	if res.Status != "conflict" {
+		t.Fatalf("after %d answered pushes: %+v, want a conflict", pushLoopRounds, res)
+	}
+	if got := env.transport.pullTriggers; got != pushLoopRounds {
+		t.Errorf("the peer was asked to pull %d times, want %d: the last answered push must not trigger again", got, pushLoopRounds)
+	}
+	if _, open := env.engine.ActiveConflicts()["game1"]; !open {
+		t.Error("no conflict recorded for the user to decide")
+	}
+	if got := readFile(t, env.localDir, "slot1.sav"); got != "a-turn" {
+		t.Errorf("this device's save became %q", got)
+	}
+	if got := readFile(t, env.remoteDir, "slot1.sav"); got != "b-turn" {
+		t.Errorf("the peer's save became %q", got)
+	}
+}
+
+// The same push again, but not as the peer's answer — this device syncing on
+// its own while the peer has not pulled yet — is a retry, not a loop.
+func TestPushLoop_RetryingOnItsOwnIsNoLoop(t *testing.T) {
+	env := aba(t)
+	for i := 0; i < pushLoopRounds+2; i++ {
+		res, err := env.engine.SyncWithPeer(context.Background(), "game1", env.peer)
+		if err != nil || res.Status != "triggered_peer_pull" {
+			t.Fatalf("own sync %d: %+v %v, want a plain push", i, res, err)
+		}
+	}
+}
+
+// A peer that changed its save since the last push did not refuse it: the
+// watch starts over.
+func TestPushLoop_APeerThatChangedStartsOver(t *testing.T) {
+	env := aba(t)
+	ctx := context.Background()
+	if res, _ := env.engine.SyncWithPeer(ctx, "game1", env.peer); res.Status != "triggered_peer_pull" {
+		t.Fatalf("setup: %+v", res)
+	}
+	env.engine.NoteTriggered("game1", env.peer.ID)
+	if res, _ := env.engine.SyncWithPeer(ctx, "game1", env.peer); res.Status != "triggered_peer_pull" {
+		t.Fatalf("first answered push: %+v", res)
+	}
+	// The peer saves again: an older file with new content, so this device
+	// still pushes, but about a different pair of states.
+	write(t, env.remoteDir, "extra.sav", "new on the peer")
+	remote, err := delta.BuildManifest(env.remoteDir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := env.store.SetAgreedHash("game1", env.peer.ID, remote.ManifestHash()); err != nil {
+		t.Fatal(err)
+	}
+	env.engine.NoteTriggered("game1", env.peer.ID)
+	res, err := env.engine.SyncWithPeer(ctx, "game1", env.peer)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.Status == "conflict" {
+		t.Errorf("a peer whose save changed was taken for one pushing back unchanged: %+v", res)
+	}
+}
+
+func readFile(t *testing.T, dir, rel string) string {
+	t.Helper()
+	b, err := os.ReadFile(filepath.Join(dir, filepath.FromSlash(rel)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return string(b)
+}

internal/p2p/wanclient_handlers.go

@@ -563,6 +563,7 @@ func (w *WanClient) routeRequest(ctx context.Context, msg RelayMessage) (int, an
 
 	case strings.HasPrefix(route, "/sync/trigger/"):
 		gameID := route[strings.LastIndex(route, "/")+1:]
+		w.engine.Sync.NoteTriggered(gameID, from) // syncengine/pushloop.go
 		w.engine.GoSync(func(ctx context.Context) {
 			syncCtx, cancel := context.WithTimeout(ctx, 10*time.Minute)
 			defer cancel()