diff --git a/e2e/quick_turns_test.go b/e2e/quick_turns_test.go new file mode 100644 index 00000000..22ddc76c --- /dev/null +++ b/e2e/quick_turns_test.go @@ -0,0 +1,43 @@ +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"}) + 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 + } +} diff --git a/internal/p2p/routes.go b/internal/p2p/routes.go index 67d11a5c..8a1641ff 100644 --- a/internal/p2p/routes.go +++ b/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. diff --git a/internal/p2p/syncengine/agreed_report_test.go b/internal/p2p/syncengine/agreed_report_test.go new file mode 100644 index 00000000..bc538b2d --- /dev/null +++ b/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) + } +} diff --git a/internal/p2p/syncengine/engine.go b/internal/p2p/syncengine/engine.go index f60d67fe..b88759eb 100644 --- a/internal/p2p/syncengine/engine.go +++ b/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) +} diff --git a/internal/p2p/syncengine/engine_test.go b/internal/p2p/syncengine/engine_test.go index 1ed654ff..286223aa 100644 --- a/internal/p2p/syncengine/engine_test.go +++ b/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) } } diff --git a/internal/p2p/wanclient_handlers.go b/internal/p2p/wanclient_handlers.go index 95203dd3..e719a3e1 100644 --- a/internal/p2p/wanclient_handlers.go +++ b/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.