From 22401c02d81d2de41686a516468bad424d03b5ed Mon Sep 17 00:00:00 2001 From: ChakraFusion <112588066+ChakraFusion@users.noreply.github.com> Date: Thu, 15 Oct 2026 21:44:59 +0200 Subject: [PATCH] sync: two devices pushing to each other is a conflict, not a loop Since #42 a side whose manifest equals the agreed base counts as unchanged. When the same bytes come back while one device's base is still a round behind, that device reads the other's new save as "unchanged" and 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 while the change never arrives (#71). The rule stays; the loop is caught. A peer asks this device to pull (NoteTriggered, from the LAN and relay trigger handlers), and this device would answer with a push about the very same two states it pushed about last time: the peer was told to pull and pushed back instead, with neither side changed. After pushLoopRounds (2) such answers the sync raises a conflict through the existing path, so both saves stay as they are, the user decides, and no further trigger goes out. Anything other than a push (a pull, deletions, agreement), or either side's content changing, ends the watch; this device repeating its own push without the peer asking is a retry and never counts. TestPushLoop_TwoDevicesPushingToEachOtherIsAConflict is #71's reproduction and fails without the check; TestPushLoop_RetryingOnItsOwnIsNoLoop and TestPushLoop_APeerThatChangedStartsOver cover what must not count. #42's rollback tests are unchanged. TestReportedWAN_AlternatingEditsMustNotConflict wrote the same "a-turn" and "b-turn" every round, which made it this case by accident; each turn now writes something new, so it tests alternating edits again. Co-Authored-By: Claude Opus 5.5 --- e2e/reported_conflict_wan_test.go | 18 +++- internal/p2p/routes.go | 1 + internal/p2p/syncengine/engine.go | 17 +++ internal/p2p/syncengine/pushloop.go | 102 ++++++++++++++++++ internal/p2p/syncengine/pushloop_test.go | 127 +++++++++++++++++++++++ internal/p2p/wanclient_handlers.go | 1 + 6 files changed, 262 insertions(+), 4 deletions(-) create mode 100644 internal/p2p/syncengine/pushloop.go create mode 100644 internal/p2p/syncengine/pushloop_test.go diff --git a/e2e/reported_conflict_wan_test.go b/e2e/reported_conflict_wan_test.go index e952cd58..5f83add8 100644 --- a/e2e/reported_conflict_wan_test.go +++ b/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) } } diff --git a/internal/p2p/routes.go b/internal/p2p/routes.go index 526de11e..cccb9572 100644 --- a/internal/p2p/routes.go +++ b/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() diff --git a/internal/p2p/syncengine/engine.go b/internal/p2p/syncengine/engine.go index 7df1eda5..6dd2bc67 100644 --- a/internal/p2p/syncengine/engine.go +++ b/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) diff --git a/internal/p2p/syncengine/pushloop.go b/internal/p2p/syncengine/pushloop.go new file mode 100644 index 00000000..74511c50 --- /dev/null +++ b/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 +} diff --git a/internal/p2p/syncengine/pushloop_test.go b/internal/p2p/syncengine/pushloop_test.go new file mode 100644 index 00000000..ea4fdd42 --- /dev/null +++ b/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) +} diff --git a/internal/p2p/wanclient_handlers.go b/internal/p2p/wanclient_handlers.go index df1c61be..7c77cf87 100644 --- a/internal/p2p/wanclient_handlers.go +++ 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()