diff --git a/internal/p2p/engine.go b/internal/p2p/engine.go index b1777e2e..b0029623 100644 --- a/internal/p2p/engine.go +++ b/internal/p2p/engine.go @@ -129,6 +129,10 @@ type Engine struct { deleteRefreshMu sync.Mutex deleteRefresh map[string]bool + // reports are what a peer's state-carrying reports leave to be done once + // they are answered (see handleSyncEvent). + reports reportsRecorded + // Lifecycle of background work this engine starts. Syncs triggered by a // peer appearing, or by the retry failsafe, used to run on // context.Background() and so outlived Stop entirely — leaving a partly diff --git a/internal/p2p/reports.go b/internal/p2p/reports.go new file mode 100644 index 00000000..3b264e6b --- /dev/null +++ b/internal/p2p/reports.go @@ -0,0 +1,52 @@ +package p2p + +import ( + "net/http" + "sync" +) + +// A peer's reports that carry state ("sync-complete", "in-sync") are answered +// before what they record is done, and are sent again until they are answered +// (ReportSyncEvent). So one can arrive twice — the first time it arrived, its +// answer was what got lost — and two can be recorded at once. reportsRecorded +// does each report's recording once, and one at a time. + +// reportsRemembered is how many report IDs are kept to recognise a report +// sent again. A report is sent again within a minute, among a handful. +const reportsRemembered = 256 + +type reportsRecorded struct { + mu sync.Mutex // held while a report is recorded + seen map[string]bool + ids []string // in arrival order, to forget the oldest +} + +// once runs record for a report unless a report with the same ID already ran. +// A report without an ID, from an older build, is never sent again, and always +// runs. +func (r *reportsRecorded) once(data map[string]any, record func()) { + r.mu.Lock() + defer r.mu.Unlock() + if id, _ := data["reportId"].(string); id != "" { + if r.seen[id] { + return + } + if r.seen == nil { + r.seen = map[string]bool{} + } + r.seen[id] = true + r.ids = append(r.ids, id) + if len(r.ids) > reportsRemembered { + delete(r.seen, r.ids[0]) + r.ids = r.ids[1:] + } + } + record() +} + +// answerNow sends the report's answer at once, so the sender is done with it +// while the handler goes on to record it. +func answerNow(w http.ResponseWriter) { + jsonOK(w, map[string]any{"success": true}) + _ = http.NewResponseController(w).Flush() +} diff --git a/internal/p2p/reports_test.go b/internal/p2p/reports_test.go new file mode 100644 index 00000000..4624fef5 --- /dev/null +++ b/internal/p2p/reports_test.go @@ -0,0 +1,94 @@ +package p2p + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/go-chi/chi/v5" + + "github.com/opensave/opensave/internal/p2p/syncengine" +) + +// serveReports is B's sync-event route on a real server, every request coming +// from A's address. +func (f *raceFixture) serveReports(t *testing.T) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + r.RemoteAddr = f.peer.Address + ":40000" + rc := chi.NewRouteContext() + rc.URLParams.Add("gameId", f.gameID) + f.e.handleSyncEvent(w, r.WithContext(context.WithValue(r.Context(), chi.RouteCtxKey, rc))) + })) + t.Cleanup(srv.Close) + return srv +} + +func postReport(client *http.Client, url string, data map[string]any) error { + body, _ := json.Marshal(map[string]any{"eventType": "sync-complete", "data": data}) + resp, err := client.Post(url, "application/json", bytes.NewReader(body)) + if err != nil { + return err + } + resp.Body.Close() + return nil +} + +// A report is answered before what it records is done. Recording waits for +// the database, which a snapshot or a sync can hold for seconds, and the peer +// gave up on the answer long before: the report was lost (upstream CI #385, +// #381). Here recording is held up for as long as the test likes, and the +// answer still comes at once. +func TestAReportIsAnsweredBeforeItIsRecorded(t *testing.T) { + f := newRaceFixture(t, false) + release := make(chan struct{}) + recorded := make(chan struct{}) + prev := refreshAfterPull + refreshAfterPull = func(*Engine, string, syncengine.Peer) { + <-release + close(recorded) + } + t.Cleanup(func() { refreshAfterPull = prev }) + srv := f.serveReports(t) + + client := &http.Client{Timeout: time.Second} + err := postReport(client, srv.URL, map[string]any{"pulledFiles": []string{"slot1.sav"}, "reportId": "r1"}) + close(release) + if err != nil { + t.Fatalf("the report was not answered while it was being recorded: %v", err) + } + select { + case <-recorded: + case <-time.After(10 * time.Second): + t.Fatal("the report was answered but never recorded") + } +} + +// A report sent again because its answer was lost is recorded once: the +// sender sends it with the same ID until it is answered. One without an ID, +// from an older build, is recorded every time, as before. +func TestAReportSentAgainIsRecordedOnce(t *testing.T) { + f := newRaceFixture(t, false) + var records atomic.Int32 + prev := refreshAfterPull + refreshAfterPull = func(*Engine, string, syncengine.Peer) { records.Add(1) } + t.Cleanup(func() { refreshAfterPull = prev }) + + took := map[string]any{"pulledFiles": []string{"slot1.sav"}} + f.reportOverLAN(t, map[string]any{"pulledFiles": took["pulledFiles"], "reportId": "same"}) + f.reportOverLAN(t, map[string]any{"pulledFiles": took["pulledFiles"], "reportId": "same"}) + if n := records.Load(); n != 1 { + t.Errorf("a report sent twice was recorded %d times, want once", n) + } + f.reportOverLAN(t, map[string]any{"pulledFiles": took["pulledFiles"], "reportId": "other"}) + f.reportOverLAN(t, took) + f.reportOverLAN(t, took) + if n := records.Load(); n != 4 { + t.Errorf("recorded %d times in all, want 4: another report, and two without an ID", n) + } +} diff --git a/internal/p2p/routes.go b/internal/p2p/routes.go index 526de11e..fa38c9df 100644 --- a/internal/p2p/routes.go +++ b/internal/p2p/routes.go @@ -924,20 +924,33 @@ func (e *Engine) handleSyncEvent(w http.ResponseWriter, r *http.Request) { if e.Sync.Progress.OnSyncComplete != nil { e.Sync.Progress.OnSyncComplete(gameID, ev) } - if peer, ok := e.peerByAddress(clientIP(r)); ok { - e.peerFinishedPulling(gameID, peer, body.Data) - } + // Answered first, recorded after: what it records waits for the + // database, which a snapshot or sync here can hold for seconds, and + // the peer gives up on an answer long before that (see + // ReportSyncEvent). Still recorded before this returns. + answerNow(w) + e.reports.once(body.Data, func() { + if peer, ok := e.peerByAddress(clientIP(r)); ok { + e.peerFinishedPulling(gameID, peer, body.Data) + } + }) + return case "in-sync": // The peer verified both sides hold identical content; confirm on // our side (hash re-check) before recording lineage + last-synced. + // Answered first, as above. claimedHash, _ := body.Data["manifestHash"].(string) - if peer, ok := e.peerByAddress(clientIP(r)); ok { - go func() { - ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) - defer cancel() - e.Sync.ConfirmInSync(ctx, gameID, peer, claimedHash) - }() - } + answerNow(w) + e.reports.once(body.Data, func() { + if peer, ok := e.peerByAddress(clientIP(r)); ok { + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + e.Sync.ConfirmInSync(ctx, gameID, peer, claimedHash) + }() + } + }) + return case "sync-error": if e.Sync.Progress.OnSyncError != nil { e.Sync.Progress.OnSyncError(gameID, ev) diff --git a/internal/p2p/transport_lan.go b/internal/p2p/transport_lan.go index 1955b640..11f31907 100644 --- a/internal/p2p/transport_lan.go +++ b/internal/p2p/transport_lan.go @@ -3,6 +3,8 @@ package p2p import ( "bytes" "context" + "crypto/rand" + "encoding/hex" "encoding/json" "fmt" "io" @@ -115,6 +117,35 @@ const eventQueueSize = 64 // it stops, so a peer that has gone away does not keep a goroutine forever. const eventQueueIdle = time.Minute +// How a report that carries state is sent: each try given stateReportTimeout, +// then a pause of stateReportBackoff, doubling, between tries — about half a +// minute in all before it is given up. Variables so a test can shorten them. +var ( + stateReportTimeout = 5 * time.Second + stateReportBackoff = time.Second +) + +const stateReportTries = 4 + +// carriesState says whether a report tells the other device something it +// records — what both devices now hold — rather than how a sync is going. +func carriesState(eventType string) bool { + return eventType == "sync-complete" || eventType == "in-sync" +} + +// withReportID is the report's data with an ID of its own added, the same for +// every send of it (see ReportSyncEvent). The caller's map is not changed. +func withReportID(data map[string]any) map[string]any { + out := make(map[string]any, len(data)+1) + for k, v := range data { + out[k] = v + } + var id [8]byte + _, _ = rand.Read(id[:]) + out["reportId"] = hex.EncodeToString(id[:]) + return out +} + // ReportSyncEvent tells the peer how a sync is going, without waiting for it. // // In order, through one sender per peer. Each report used to go out on its @@ -122,19 +153,46 @@ const eventQueueIdle = time.Minute // the "finished": the peer then showed a sync running that was already over — // in its transfers list, and on the game's card — until something timed it // out. +// +// A lost progress report costs nothing, and is sent once. A report that +// carries state (carriesState) is sent again until it is answered: sent once +// with a short limit, it was lost whenever the other device was busy for a +// moment, which left that device's record of what both hold a round behind, +// and the next save read as both sides having changed — a false conflict +// (upstream CI #385). Later reports to the peer wait meanwhile, so the order +// holds. Each send carries the same reportId, so a report that did arrive, +// and whose answer was what got lost, is recorded once. func (t *lanTransport) ReportSyncEvent(peer syncengine.Peer, gameID, eventType string, data map[string]any) { + stateful := carriesState(eventType) + if stateful { + data = withReportID(data) + } post := func() { - ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) - defer cancel() - err := t.postJSON(ctx, peer, peerURL(peer, "/sync-event/"+gameID), map[string]any{ - "eventType": eventType, "data": data, - }, nil) - // A lost progress report costs nothing. These two carry state — what - // the other device records as synced and agreed — and losing one - // was invisible on both sides. - if err != nil && (eventType == "in-sync" || eventType == "sync-complete") && t.engine != nil && t.engine.Log != nil { - t.engine.Log("info", fmt.Sprintf("could not tell %s that %s is %s: %v", - peer.Name, gameID, eventType, err)) + send := func(limit time.Duration) error { + ctx, cancel := context.WithTimeout(context.Background(), limit) + defer cancel() + return t.postJSON(ctx, peer, peerURL(peer, "/sync-event/"+gameID), map[string]any{ + "eventType": eventType, "data": data, + }, nil) + } + if !stateful { + _ = send(2 * time.Second) + return + } + var err error + wait := stateReportBackoff + for try := 1; try <= stateReportTries; try++ { + if err = send(stateReportTimeout); err == nil { + return + } + if try < stateReportTries { + time.Sleep(wait) + wait *= 2 + } + } + if t.engine != nil && t.engine.Log != nil { + t.engine.Log("info", fmt.Sprintf("could not tell %s that %s is %s, after %d tries: %v", + peer.Name, gameID, eventType, stateReportTries, err)) } } t.eventsMu.Lock() diff --git a/internal/p2p/transport_lan_events_test.go b/internal/p2p/transport_lan_events_test.go index 54bcaf98..18295b68 100644 --- a/internal/p2p/transport_lan_events_test.go +++ b/internal/p2p/transport_lan_events_test.go @@ -65,3 +65,75 @@ func TestSyncEventReportsArriveInOrder(t *testing.T) { } } } + +// A report that carries state is not lost to one slow answer. "sync-complete" +// tells the other device what both now hold; sent once, with a short limit, a +// device busy for a moment missed it, kept a base a round behind, and read +// the next save as both sides having changed (the Journey test's false +// conflict, upstream CI #385). It is sent again, with the same ID, until it is +// answered; a progress report is still sent once. +func TestSyncEventReportsThatCarryStateAreSentAgain(t *testing.T) { + short(t, &stateReportTimeout, 200*time.Millisecond) + short(t, &stateReportBackoff, 50*time.Millisecond) + + var mu sync.Mutex + var got []string + ids := map[string]bool{} + slowFirst := map[string]bool{"sync-complete": true, "sync-progress": true} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body struct { + EventType string `json:"eventType"` + Data map[string]any `json:"data"` + } + _ = json.NewDecoder(r.Body).Decode(&body) + mu.Lock() + slow := slowFirst[body.EventType] + delete(slowFirst, body.EventType) + got = append(got, body.EventType) + if id, _ := body.Data["reportId"].(string); id != "" { + ids[id] = true + } + mu.Unlock() + if slow { + time.Sleep(2500 * time.Millisecond) // busy: answered after any limit used + } + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + host, portStr, _ := net.SplitHostPort(srv.Listener.Addr().String()) + port, _ := strconv.Atoi(portStr) + peer := syncengine.Peer{ID: "peer1", Name: "Deck", Address: host, Port: port} + + tr := &lanTransport{} + tr.ReportSyncEvent(peer, "hades", "sync-progress", nil) + tr.ReportSyncEvent(peer, "hades", "sync-complete", map[string]any{"pulledFiles": []string{"a.sav"}}) + tr.ReportSyncEvent(peer, "hades", "e-after", nil) + + deadline := time.Now().Add(10 * time.Second) + for { + mu.Lock() + n := len(got) + mu.Unlock() + if n >= 4 || time.Now().After(deadline) { + break + } + time.Sleep(10 * time.Millisecond) + } + mu.Lock() + defer mu.Unlock() + want := []string{"sync-progress", "sync-complete", "sync-complete", "e-after"} + if fmt.Sprint(got) != fmt.Sprint(want) { + t.Fatalf("received %v, want %v: the slow sync-complete sent again, in order; the progress report once", got, want) + } + if len(ids) != 1 { + t.Errorf("the two sends of one report carried %d IDs, want the same one", len(ids)) + } +} + +// short lowers a timing for one test. +func short(t *testing.T, v *time.Duration, d time.Duration) { + t.Helper() + prev := *v + *v = d + t.Cleanup(func() { *v = prev }) +}