From 8fc74c908a5dc47131cc0dc60f3118052ac55e3c Mon Sep 17 00:00:00 2001 From: John O'Keefe Date: Sat, 17 Oct 2026 10:26:22 -0400 Subject: [PATCH] =?UTF-8?q?fix(conflicts):=20resolution=20writes=20funnel?= =?UTF-8?q?=20through=20SaveProgress=20=E2=80=94=20audio=20position=20surv?= =?UTF-8?q?ives?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Batch H1 step 3b — THE data fix. Conflict resolution was the one live erasure writer: applyProgressResolution and its verbatim twin applyResolution both wrote the winner via a raw UpdateUniversalProgress whose param list omits audio_offset_seconds/context_text/viewports, and the SQL's DO UPDATE SET ... = EXCLUDED... NULLed exactly those. The web UI resolves through POST /api/conflicts/:id/resolve and /api/conflicts/bulk-resolve (main.js resolveConflict/bulkResolve), and the reader's mark-as-read fires a resolution first when the reader SSR stamps data-conflict-id/-winner — so every resolution today erased the audiobook position and reader context. Now: ProgressService is a required ConflictHandler dependency; BOTH call sites (single resolve + bulk resolve) share one applyProgressResolution that funnels through SaveProgress (the verbatim twin is deleted). Deliberate choices, per the plan: - SkipConflictDetection on the resolution write: both flows apply BEFORE the conflict row is marked resolved, so the funnel's LWW detector would otherwise file a duplicate pending conflict (HasRecentConflictResolution is false at apply time). - Source "conflict_resolution" — the row's last-write attribution; the winner stays durably recorded in resolution_data.winner. - Broadcast: true — live devices jump to the winning position. - The winner snapshot maps percentage/epubcfi/context_text/chapter/ character/page/total_pages; a winner without a CFI clears the locator pair (ClearLocator) instead of re-pairing with the loser's stale anchor. Everything the snapshot doesn't carry SURVIVES via the LWW seed: audio_offset_seconds above all. chapter_progress keeps its stored value (was overwritten with the book percentage), an absent percentage survives (was written as 0%). - Accepted: the resolution captures one reading_history session (attributed to conflict_resolution; the 2h idle ceiling applies). Verification: TestH1_ConflictResolutionPreservesAudio drives the real flow — web push with audio 300s, diverging push from a second source detects a conflict, resolve with the second source as winner, GET shows percentage 0.3 AND audio_offset_seconds 300 (previously NULLed), and no duplicate pending conflict remains. Full contract pin suite green. The host-run TestConflict*/TestProgress* failures are environmental (container-only /app/uploads path + db hostname; fail identically at baseline b4ae423) — the same suites pass in the test-runner container. --- cmd/server/main.go | 2 +- cmd/server/tests/contract_pins_test.go | 105 ++++++++++++++- cmd/server/tests/test_helpers_test.go | 2 +- internal/handlers/conflicts.go | 171 ++++++++----------------- internal/sync/progress.go | 10 +- 5 files changed, 166 insertions(+), 124 deletions(-) diff --git a/cmd/server/main.go b/cmd/server/main.go index 30ba500b..1168c488 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -151,7 +151,7 @@ func main() { koreaderHandler.SetAnnotationService(annotationService) koreaderHandler.SetLibraryService(libraryService) wsHandler := handlers.NewWSHandler(queries, connManager, cfg.JWTSecret, deviceAuthMiddleware) - conflictHandler := handlers.NewConflictHandler(queries, connManager) + conflictHandler := handlers.NewConflictHandler(queries, connManager, progressService) analyticsHandler := handlers.NewAnalyticsHandler(queries) statisticsService := services.NewStatisticsService(queries) statsHandler := handlers.NewStatsHandler(statisticsService) diff --git a/cmd/server/tests/contract_pins_test.go b/cmd/server/tests/contract_pins_test.go index 43ec9e2d..ede68c95 100644 --- a/cmd/server/tests/contract_pins_test.go +++ b/cmd/server/tests/contract_pins_test.go @@ -146,11 +146,19 @@ func createContractFixture(t *testing.T, setup *TestServerSetup) contractFixture // (testuser) and returns its auth token. Bearer-token auth for the // KOReader endpoints, URL-token auth for the Kobo endpoints. func createContractDevice(t *testing.T, setup *TestServerSetup, deviceType string) string { + return createContractDeviceNamed(t, setup, "Test Contract Device", deviceType) +} + +// createContractDeviceNamed registers a device with an explicit name — +// the device NAME is what the web progress push adopts as its sync +// source (X-Device-Token attribution), so conflict tests can drive a +// second named source without a separate account. +func createContractDeviceNamed(t *testing.T, setup *TestServerSetup, name, deviceType string) string { t.Helper() token := fmt.Sprintf("dev_contract_%s", uuid.New().String()) device, err := setup.DB.CreateDevice(context.Background(), database.CreateDeviceParams{ UserID: pgtype.UUID{Bytes: getTestUserID(t, setup.DB), Valid: true}, - DeviceName: "Test Contract Device", + DeviceName: name, DeviceType: deviceType, DeviceIdentifier: fmt.Sprintf("contract-%s-%d", deviceType, time.Now().UnixNano()), AuthToken: token, @@ -822,3 +830,98 @@ func TestContractPin_SSRPages(t *testing.T) { assert.Contains(t, html, "- Bookhoard") }) } + +// --------------------------------------------------------------------------- +// 6. H1 verification — resolving a conflict must preserve unknown fields +// --------------------------------------------------------------------------- + +// TestH1_ConflictResolutionPreservesAudio is the executable check for the +// hardening plan's H1 done-when: a resolution applied via +// POST /api/conflicts/:id/resolve (the endpoint the web UI and the +// reader's mark-as-read both use) must NOT erase fields the winner's +// snapshot doesn't carry (before the funnel, the resolution write NULLed +// audio_offset_seconds), and the funneled resolution write must not file +// a duplicate pending conflict on the item it just resolved. +func TestH1_ConflictResolutionPreservesAudio(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + setup := setupTestServer(t) + fx := createContractFixture(t, setup) + progress := progressURL(setup.Server.URL, fx.itemID) + + // Push 1: web push carrying an audiobook-style position (source "web") + // — audio_offset_seconds is stored but NOT part of conflict snapshots. + resp := doReq(t, "PUT", progress, map[string]interface{}{ + "percentage": 0.5, + "audio_offset_seconds": 300, + }, setup.Token) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + + // Push 2 within the 5-minute window from a second source (a + // device-token push attributes the write to the device's name) with a + // diverging percentage and NO audio field: the funnel detects the + // source mismatch + pct diff and records a conflict. + deviceToken := createContractDeviceNamed(t, setup, "koreader", "koreader") + push2Req, err := http.NewRequest("PUT", progress, + strings.NewReader(`{"percentage": 0.3}`)) + require.NoError(t, err) + push2Req.Header.Set("Content-Type", "application/json") + push2Req.Header.Set("Authorization", "Bearer "+setup.Token) + push2Req.Header.Set("X-Device-Token", deviceToken) + resp2, err := (&http.Client{}).Do(push2Req) + require.NoError(t, err) + require.Equal(t, http.StatusOK, resp2.StatusCode) + resp2.Body.Close() + + // Find the freshly detected progress conflict. + listResp := doReq(t, "GET", setup.Server.URL+"/api/conflicts", nil, setup.Token) + defer listResp.Body.Close() + require.Equal(t, http.StatusOK, listResp.StatusCode) + list := decodeJSON(t, listResp) + conflicts, ok := list["conflicts"].([]interface{}) + require.True(t, ok, "conflicts should be an array") + var conflictID string + for _, raw := range conflicts { + col := raw.(map[string]interface{}) + if col["media_item_id"] == fx.itemID && col["conflict_type"] == "progress" && col["resolution_status"] == "unresolved" { + conflictID, _ = col["id"].(string) + } + } + require.NotEmpty(t, conflictID, "the two-source push should have detected a progress conflict") + + // Resolve with the second source as the winner — the endpoint and + // payload shape the web UI posts (web/static/main.js resolveConflict). + resolve := doReq(t, "POST", setup.Server.URL+"/api/conflicts/"+conflictID+"/resolve", map[string]interface{}{ + "winner": "koreader", + }, setup.Token) + defer resolve.Body.Close() + require.Equal(t, http.StatusOK, resolve.StatusCode) + resolved := decodeJSON(t, resolve) + require.Equal(t, true, resolved["conflict_resolved"]) + + // The saved row carries the winner's percentage AND the audio field + // the winner never knew about. + get := doReq(t, "GET", progress, nil, setup.Token) + defer get.Body.Close() + require.Equal(t, http.StatusOK, get.StatusCode) + row := decodeJSON(t, get) + require.Equal(t, 0.3, getFloatField(t, row, "percentage")) + require.Equal(t, float64(300), getFloatField(t, row, "audio_offset_seconds"), + "resolution must preserve audio_offset_seconds (LWW seed), not NULL it") + + // No duplicate pending conflict may remain for this item (the funnel's + // own LWW detector must skip the resolution write). + get2 := doReq(t, "GET", setup.Server.URL+"/api/conflicts", nil, setup.Token) + defer get2.Body.Close() + require.Equal(t, http.StatusOK, get2.StatusCode) + list2 := decodeJSON(t, get2) + for _, raw := range list2["conflicts"].([]interface{}) { + col := raw.(map[string]interface{}) + if col["media_item_id"] == fx.itemID && col["conflict_type"] == "progress" { + require.Equal(t, "user_resolved", col["resolution_status"], + "the resolved conflict must not have spawned a duplicate pending conflict") + } + } +} diff --git a/cmd/server/tests/test_helpers_test.go b/cmd/server/tests/test_helpers_test.go index c10de780..7bee6a77 100644 --- a/cmd/server/tests/test_helpers_test.go +++ b/cmd/server/tests/test_helpers_test.go @@ -490,7 +490,7 @@ func setupTestServer(t *testing.T) *TestServerSetup { koreaderHandler := handlers.NewKOReaderHandler(queries, connManager, queueProcessor, progressService) koreaderHandler.SetAnnotationService(annotationService) wsHandler := handlers.NewWSHandler(queries, connManager, cfg.JWTSecret, deviceAuthMiddleware) - conflictHandler := handlers.NewConflictHandler(queries, connManager) + conflictHandler := handlers.NewConflictHandler(queries, connManager, progressService) analyticsHandler := handlers.NewAnalyticsHandler(queries) queueHandler := handlers.NewQueueHandler(queries, queueProcessor) systemSettingsHandler := handlers.NewSystemSettingsHandler(queries) diff --git a/internal/handlers/conflicts.go b/internal/handlers/conflicts.go index 7ba495c9..6d2bd2cd 100644 --- a/internal/handlers/conflicts.go +++ b/internal/handlers/conflicts.go @@ -18,12 +18,18 @@ import ( type ConflictHandler struct { db *database.Queries connManager *wsync.ConnectionManager + progressSvc *wsync.ProgressService } -func NewConflictHandler(db *database.Queries, connManager *wsync.ConnectionManager) *ConflictHandler { +// NewConflictHandler requires the shared ProgressService: applying a +// resolution is a progress WRITE and must funnel through SaveProgress so +// the winner's snapshot lands without erasing row fields it doesn't carry +// (audio_offset_seconds survives the write). +func NewConflictHandler(db *database.Queries, connManager *wsync.ConnectionManager, progressSvc *wsync.ProgressService) *ConflictHandler { return &ConflictHandler{ db: db, connManager: connManager, + progressSvc: progressSvc, } } @@ -294,71 +300,64 @@ func (h *ConflictHandler) ResolveConflict(c *echo.Context) error { return c.JSON(http.StatusOK, response) } +// applyProgressResolution writes the winning snapshot through the +// SaveProgress funnel. Deliberate choices (hardening H1): +// - Source "conflict_resolution": the row's last-write attribution; the +// winner's identity is durably recorded on the conflict row +// (resolution_data.winner). +// - SkipConflictDetection: both flows apply BEFORE the conflict row is +// marked resolved, so the funnel's LWW detector would otherwise file a +// duplicate pending conflict on the item just resolved +// (HasRecentConflictResolution is false at apply time). +// - Broadcast: live devices jump to the winning position, same as a +// device push; the write also captures one reading_history session +// (idle-gap ceiling applies) — accepted as the resolution's own event. +// - Every field the winner's snapshot doesn't carry (audio_offset_seconds, +// viewports, zoom, reading_mode, and an absent percentage) SURVIVES via +// the service's LWW seed instead of being NULLed — that is the fix. +// chapter_progress keeps its stored value instead of being overwritten +// with the book-level percentage. func (h *ConflictHandler) applyProgressResolution(mediaItemID pgtype.UUID, userID pgtype.UUID, winnerSource string, data map[string]interface{}) error { - ctx := context.Background() - - existingProgress, err := h.db.GetReadingProgress(ctx, database.GetReadingProgressParams{ - MediaItemID: mediaItemID, - UserID: userID, - }) - if err != nil && !errors.Is(err, pgx.ErrNoRows) { - return err + req := wsync.SaveProgressRequest{ + MediaItemID: mediaItemID, + UserID: userID, + Source: "conflict_resolution", + SkipConflictDetection: true, + Broadcast: true, + // A winner without a CFI (percentage-only sources, e.g. Kobo) + // declares no locator: clear the inherited anchor pair instead of + // re-pairing the new percentage with the loser's stale CFI/ + // context. A winner WITH a CFI keeps the stored pair coherent. + ClearLocator: true, } - - percentage := 0.0 if p, ok := data["percentage"].(float64); ok { - percentage = p + req.Percentage = &p } - - var epubcfi pgtype.Text if e, ok := data["epubcfi"].(string); ok { - epubcfi = pgtype.Text{String: e, Valid: true} + req.Epubcfi = &e + req.ClearLocator = false + } + if ct, ok := data["context_text"].(string); ok { + req.ContextText = &ct } - - var chapter pgtype.Int4 if c, ok := data["chapter"].(float64); ok { - chapter = pgtype.Int4{Int32: int32(c), Valid: true} + ch := int(c) + req.Chapter = &ch } - - var characterOffset pgtype.Int8 if c, ok := data["character"].(float64); ok { - characterOffset = pgtype.Int8{Int64: int64(c), Valid: true} + co := int64(c) + req.CharacterOffset = &co } - - var currentPage pgtype.Int4 - var totalPages pgtype.Int4 - if err == nil { - currentPage = existingProgress.CurrentPage - totalPages = existingProgress.TotalPages - } - if p, ok := data["page"].(float64); ok { - currentPage = pgtype.Int4{Int32: int32(p), Valid: true} + pg := int(p) + req.CurrentPage = &pg } if p, ok := data["total_pages"].(float64); ok { - totalPages = pgtype.Int4{Int32: int32(p), Valid: true} + tp := int(p) + req.TotalPages = &tp } - _, err = h.db.UpdateUniversalProgress(ctx, database.UpdateUniversalProgressParams{ - MediaItemID: mediaItemID, - UserID: userID, - Percentage: pgtype.Float8{Float64: percentage, Valid: true}, - Epubcfi: epubcfi, - Chapter: chapter, - ChapterProgress: pgtype.Float8{Float64: percentage, Valid: true}, - CharacterOffset: characterOffset, - CurrentPage: currentPage, - TotalPages: totalPages, - LastSyncDevice: pgtype.Text{String: "conflict_resolution", Valid: true}, - LastSyncSource: pgtype.Text{String: winnerSource, Valid: true}, - ViewportY: pgtype.Float8{}, - ScrollPositionX: pgtype.Float8{}, - ScrollPositionY: pgtype.Float8{}, - PanelNumber: pgtype.Int4{}, - ReadingMode: pgtype.Text{}, - ZoomLevel: pgtype.Float8{}, - }) - + _, err := h.progressSvc.SaveProgress(context.Background(), req) return err } @@ -699,7 +698,7 @@ func (h *ConflictHandler) BulkResolveConflicts(c *echo.Context) error { continue } - if err := h.applyResolution(conflict.MediaItemID, conflict.UserID, winningSource, winnerData); err != nil { + if err := h.applyProgressResolution(conflict.MediaItemID, conflict.UserID, winningSource, winnerData); err != nil { results = append(results, ConflictResult{ ConflictID: conflictIDStr, Status: "error", @@ -783,74 +782,6 @@ func (h *ConflictHandler) getHighestProgressSource(conflictData map[string]Confl return highestSource, highestData } -func (h *ConflictHandler) applyResolution(mediaItemID pgtype.UUID, userID pgtype.UUID, winnerSource string, data map[string]interface{}) error { - ctx := context.Background() - - existingProgress, err := h.db.GetReadingProgress(ctx, database.GetReadingProgressParams{ - MediaItemID: mediaItemID, - UserID: userID, - }) - if err != nil && !errors.Is(err, pgx.ErrNoRows) { - return err - } - - percentage := 0.0 - if p, ok := data["percentage"].(float64); ok { - percentage = p - } - - var epubcfi pgtype.Text - if e, ok := data["epubcfi"].(string); ok { - epubcfi = pgtype.Text{String: e, Valid: true} - } - - var chapter pgtype.Int4 - if c, ok := data["chapter"].(float64); ok { - chapter = pgtype.Int4{Int32: int32(c), Valid: true} - } - - var characterOffset pgtype.Int8 - if c, ok := data["character"].(float64); ok { - characterOffset = pgtype.Int8{Int64: int64(c), Valid: true} - } - - var currentPage pgtype.Int4 - var totalPages pgtype.Int4 - if err == nil { - currentPage = existingProgress.CurrentPage - totalPages = existingProgress.TotalPages - } - - if p, ok := data["page"].(float64); ok { - currentPage = pgtype.Int4{Int32: int32(p), Valid: true} - } - if p, ok := data["total_pages"].(float64); ok { - totalPages = pgtype.Int4{Int32: int32(p), Valid: true} - } - - _, err = h.db.UpdateUniversalProgress(ctx, database.UpdateUniversalProgressParams{ - MediaItemID: mediaItemID, - UserID: userID, - Percentage: pgtype.Float8{Float64: percentage, Valid: true}, - Epubcfi: epubcfi, - Chapter: chapter, - ChapterProgress: pgtype.Float8{Float64: percentage, Valid: true}, - CharacterOffset: characterOffset, - CurrentPage: currentPage, - TotalPages: totalPages, - LastSyncDevice: pgtype.Text{String: "conflict_resolution", Valid: true}, - LastSyncSource: pgtype.Text{String: winnerSource, Valid: true}, - ViewportY: pgtype.Float8{}, - ScrollPositionX: pgtype.Float8{}, - ScrollPositionY: pgtype.Float8{}, - PanelNumber: pgtype.Int4{}, - ReadingMode: pgtype.Text{}, - ZoomLevel: pgtype.Float8{}, - }) - - return err -} - func (h *ConflictHandler) BulkDismissConflicts(c *echo.Context) error { user := c.Get("user").(database.Users) diff --git a/internal/sync/progress.go b/internal/sync/progress.go index 97213f41..ceed3ae7 100644 --- a/internal/sync/progress.go +++ b/internal/sync/progress.go @@ -341,6 +341,14 @@ type SaveProgressRequest struct { // percentage restores are approximate, and the next anchor-bearing // write re-establishes the pair. Request-supplied fields always win. ClearLocator bool `json:"-"` + + // SkipConflictDetection declares this write APPLIES a conflict + // resolution rather than continuing the sync stream: the two flows + // (single + bulk) apply the winner BEFORE the conflict row is marked + // resolved, so the funnel's LWW detector would see the losing source + // within the 5-minute window and file a duplicate pending conflict on + // the item the user just resolved. Server-owned (json:"-"). + SkipConflictDetection bool `json:"-"` } func (s *ProgressService) SaveProgress(ctx context.Context, req SaveProgressRequest) (database.ReadingProgress, error) { @@ -529,7 +537,7 @@ func (s *ProgressService) SaveProgress(ctx context.Context, req SaveProgressRequ } conflictDetected := false - if hasExisting && existing.LastSyncSource.Valid && existing.LastSyncSource.String != req.Source { + if !req.SkipConflictDetection && hasExisting && existing.LastSyncSource.Valid && existing.LastSyncSource.String != req.Source { if existing.LastSyncTimestamp.Valid { timeDiff := time.Since(existing.LastSyncTimestamp.Time) if timeDiff < 5*time.Minute {