sync: a report of what both devices hold is answered at once and not lost

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

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()
+}

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)
+	}
+}

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)

internal/p2p/transport_lan.go

@@ -3,6 +3,8 @@ package p2p
 import (
 	"bytes"
 	"context"
+	"crypto/rand"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -115,26 +117,82 @@ 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
 // own goroutine, so for a quick sync the "started" could reach the peer after
 // 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()

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 })
+}