diff --git a/cmd/server/tests/contract_pins_test.go b/cmd/server/tests/contract_pins_test.go index ede68c95..41e1ec18 100644 --- a/cmd/server/tests/contract_pins_test.go +++ b/cmd/server/tests/contract_pins_test.go @@ -831,10 +831,74 @@ func TestContractPin_SSRPages(t *testing.T) { }) } +// TestH1_SyncQueueFlush_LandsAndPreservesAudio proves the offline flush +// end-to-end: a pending sync_queue row (what the app's checkpoint sync +// enqueues) must be picked up by the running processor, funnel through +// SaveProgress (winner percentage lands), and NOT erase row fields the +// queue snapshot doesn't carry (audio_offset_seconds survives). +// NOTE: before 2026-10-17 this test could not pass — the processor's +// enumerator passed a NULL device_id into `device_id = $1`, which matches +// no rows, so the flush loop never ran. +func TestH1_SyncQueueFlush_LandsAndPreservesAudio(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + setup := setupTestServer(t) + fx := createContractFixture(t, setup) + deviceToken := createContractDeviceNamed(t, setup, "koreader", "koreader") + + // Seed the stored position with audio via the web path. + put := doReq(t, "PUT", progressURL(setup.Server.URL, fx.itemID), map[string]interface{}{ + "percentage": 0.5, + "audio_offset_seconds": 900, + }, setup.Token) + defer put.Body.Close() + require.Equal(t, http.StatusOK, put.StatusCode) + + // Enqueue a checkpoint-style progress entry (percentage only, no + // audio — exactly what buildProgressSnapshot captures). + itemUUID, perr := uuid.Parse(fx.itemID) + require.NoError(t, perr) + deviceRow, derr := setup.DB.GetDeviceByAuthToken(context.Background(), deviceToken) + require.NoError(t, derr) + syncData, _ := json.Marshal(map[string]interface{}{"percentage": 0.8}) + _, qerr := setup.DB.CreateSyncQueueItem(context.Background(), database.CreateSyncQueueItemParams{ + DeviceID: deviceRow.ID, + MediaItemID: pgtype.UUID{Bytes: itemUUID, Valid: true}, + SyncType: "progress", + SyncData: syncData, + Priority: pgtype.Int4{Int32: 5, Valid: true}, + MaxAttempts: pgtype.Int4{Int32: 3, Valid: true}, + Status: pgtype.Text{String: "pending", Valid: true}, + }) + require.NoError(t, qerr) + + // The processor ticks every 5s; poll up to 15s for the flush. + var row map[string]interface{} + landed := false + for i := 0; i < 30 && !landed; i++ { + time.Sleep(500 * time.Millisecond) + get := doReq(t, "GET", progressURL(setup.Server.URL, fx.itemID), nil, setup.Token) + if get.StatusCode != http.StatusOK { + get.Body.Close() + continue + } + row = decodeJSON(t, get) + get.Body.Close() + if pct, ok := row["percentage"].(float64); ok && pct > 0.79 { + landed = true + } + } + require.True(t, landed, "the queued progress (0.8) should flush onto the stored row") + require.Equal(t, float64(900), getFloatField(t, row, "audio_offset_seconds"), + "the flush must preserve audio_offset_seconds (LWW seed), not NULL it") +} + // --------------------------------------------------------------------------- -// 6. H1 verification — resolving a conflict must preserve unknown fields +// 7. 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 diff --git a/cmd/server/tests/sync_integration_test.go b/cmd/server/tests/sync_integration_test.go index f0228c70..652e4513 100644 --- a/cmd/server/tests/sync_integration_test.go +++ b/cmd/server/tests/sync_integration_test.go @@ -192,7 +192,7 @@ func TestSyncQueueProcessor(t *testing.T) { // Verify queue item was created items, err := db.ListPendingSyncQueueItems(ctx, database.ListPendingSyncQueueItemsParams{ DeviceID: deviceID, - Limit: int32(10), + RowLimit: int32(10), }) require.NoError(t, err) assert.Equal(t, 1, len(items), "Should have one queue item") @@ -235,7 +235,7 @@ func TestSyncConflictDetection(t *testing.T) { // Verify both exist (conflict detection would happen during processing) items, err := db.ListPendingSyncQueueItems(ctx, database.ListPendingSyncQueueItemsParams{ DeviceID: deviceID, - Limit: int32(10), + RowLimit: int32(10), }) require.NoError(t, err) assert.Equal(t, 1, len(items), "Should have one queue item") diff --git a/internal/database/queries.sql.go b/internal/database/queries.sql.go index 0df46f52..f46ac7b5 100644 --- a/internal/database/queries.sql.go +++ b/internal/database/queries.sql.go @@ -11341,18 +11341,21 @@ func (q *Queries) ListPendingHashConflicts(ctx context.Context) ([]ListPendingHa const ListPendingSyncQueueItems = `-- name: ListPendingSyncQueueItems :many SELECT id, device_id, media_item_id, sync_type, sync_data, priority, attempts, max_attempts, status, error_message, created_at, processed_at FROM sync_queue -WHERE device_id = $1 AND status = 'pending' +WHERE ($1::uuid IS NULL OR device_id = $1::uuid) AND status = 'pending' ORDER BY priority ASC, created_at ASC -LIMIT $2 +LIMIT $2::int ` type ListPendingSyncQueueItemsParams struct { DeviceID pgtype.UUID `db:"device_id" json:"device_id"` - Limit int32 `db:"limit" json:"limit"` + RowLimit int32 `db:"row_limit" json:"row_limit"` } +// DeviceID NULL = every device's pending items (the queue processor's +// enumerator passes NULL; a bare `device_id = NULL` never matches, which +// silently disabled the offline flush until 2026-10-17). func (q *Queries) ListPendingSyncQueueItems(ctx context.Context, arg ListPendingSyncQueueItemsParams) ([]SyncQueue, error) { - rows, err := q.db.Query(ctx, ListPendingSyncQueueItems, arg.DeviceID, arg.Limit) + rows, err := q.db.Query(ctx, ListPendingSyncQueueItems, arg.DeviceID, arg.RowLimit) if err != nil { return nil, err } diff --git a/internal/database/queries/queries.sql b/internal/database/queries/queries.sql index b854585f..de4d6f50 100644 --- a/internal/database/queries/queries.sql +++ b/internal/database/queries/queries.sql @@ -1744,10 +1744,13 @@ RETURNING *; SELECT * FROM sync_queue WHERE id = $1; -- name: ListPendingSyncQueueItems :many +-- DeviceID NULL = every device's pending items (the queue processor's +-- enumerator passes NULL; a bare `device_id = NULL` never matches, which +-- silently disabled the offline flush until 2026-10-17). SELECT * FROM sync_queue -WHERE device_id = $1 AND status = 'pending' +WHERE (@device_id::uuid IS NULL OR device_id = @device_id::uuid) AND status = 'pending' ORDER BY priority ASC, created_at ASC -LIMIT $2; +LIMIT @row_limit::int; -- name: UpdateSyncQueueItemStatus :one UPDATE sync_queue diff --git a/internal/handlers/dashboard_test.go b/internal/handlers/dashboard_test.go index aae45e9c..394e3017 100644 --- a/internal/handlers/dashboard_test.go +++ b/internal/handlers/dashboard_test.go @@ -46,7 +46,7 @@ func TestBuildSections_ConvertsServiceTypesToHandlerTypes(t *testing.T) { }, } - result := BuildSections(serviceSections, "") + result := BuildSections(serviceSections, "", "", map[string]string{}) assert.Equal(t, 2, len(result), "should have 2 sections") @@ -90,7 +90,7 @@ func TestBuildSections_ConvertsItemsCorrectly(t *testing.T) { }, } - result := BuildSections(serviceSections, "") + result := BuildSections(serviceSections, "", "", map[string]string{}) assert.Equal(t, 1, len(result), "should have 1 section") section := result[0] diff --git a/internal/handlers/queue.go b/internal/handlers/queue.go index 71000fa8..dcdc339d 100644 --- a/internal/handlers/queue.go +++ b/internal/handlers/queue.go @@ -115,12 +115,12 @@ func (h *QueueHandler) ListDeviceQueueItems(c *echo.Context) error { if status != "" { items, err = h.db.ListPendingSyncQueueItems(c.Request().Context(), database.ListPendingSyncQueueItemsParams{ DeviceID: device.ID, - Limit: int32(limit), + RowLimit: int32(limit), }) } else { items, err = h.db.ListPendingSyncQueueItems(c.Request().Context(), database.ListPendingSyncQueueItemsParams{ DeviceID: device.ID, - Limit: int32(limit), + RowLimit: int32(limit), }) } diff --git a/internal/sync/offline.go b/internal/sync/offline.go index 0c722e1d..ef3881e6 100644 --- a/internal/sync/offline.go +++ b/internal/sync/offline.go @@ -171,7 +171,7 @@ func (d *OfflineDetector) processQueuedItems(ctx context.Context, deviceID pgtyp items, err := d.db.ListPendingSyncQueueItems(ctx, database.ListPendingSyncQueueItemsParams{ DeviceID: deviceID, - Limit: 50, + RowLimit: 50, }) if err != nil { diff --git a/internal/sync/queue.go b/internal/sync/queue.go index a0601018..2fbef285 100644 --- a/internal/sync/queue.go +++ b/internal/sync/queue.go @@ -298,36 +298,27 @@ func (p *SyncQueueProcessor) processQueue(ctx context.Context) { } } +// getPendingItems enumerates the pending queue. DeviceID is passed NULL: +// the processor owns every device's backlog (the per-device path remains +// available to callers that pass a concrete id). One fetch per tick — the +// old loop re-fetched the same first batch (no OFFSET in the query) and +// would have looped forever once the backlog hit the batch size. func (p *SyncQueueProcessor) getPendingItems(ctx context.Context) []SyncQueueItem { var allItems []SyncQueueItem - var offset int32 = 0 - batchSize := int32(p.batchSize) - for { - dbItems, err := p.db.ListPendingSyncQueueItems(ctx, database.ListPendingSyncQueueItemsParams{ - DeviceID: pgtype.UUID{Valid: false}, - Limit: batchSize, - }) - if err != nil { - log.Printf("Error listing pending items: %v", err) - break - } + dbItems, err := p.db.ListPendingSyncQueueItems(ctx, database.ListPendingSyncQueueItemsParams{ + DeviceID: pgtype.UUID{Valid: false}, // NULL = all devices + RowLimit: int32(p.batchSize), + }) + if err != nil { + log.Printf("Error listing pending items: %v", err) + return nil + } - if len(dbItems) == 0 { - break + for _, item := range dbItems { + if item.Status.String == SyncStatusPending { + allItems = append(allItems, dbItemToSyncQueueItem(item)) } - - for _, item := range dbItems { - if item.Status.String == SyncStatusPending { - allItems = append(allItems, dbItemToSyncQueueItem(item)) - } - } - - if int32(len(dbItems)) < batchSize { - break - } - - offset += batchSize } return allItems @@ -653,6 +644,7 @@ func dbItemToSyncQueueItem(item database.SyncQueue) SyncQueueItem { DeviceID: item.DeviceID, MediaItemID: item.MediaItemID, SyncType: item.SyncType, + SyncData: item.SyncData, // was dropped: every flushed item lost its payload Priority: item.Priority.Int32, Attempts: item.Attempts.Int32, MaxAttempts: item.MaxAttempts.Int32,