fix(sync): the offline queue flush never processed anything — fix the enumerator + the dropped payload

Found while verifying H1's "app offline-play -> flush -> position lands"
with the emulator path: the flush has been dead code in two independent
ways, invisible because the first break made the second unreachable.

1. ListPendingSyncQueueItems matched `device_id = $1` with a NULL $1
   (the processor passes DeviceID{Valid:false} to enumerate all devices;
   uuid = NULL matches no rows). The ticker therefore processed nothing,
   ever. The WHERE now treats NULL as "every device" (the same
   narg::uuid IS NULL idiom the visibility queries use; EXPLAIN'd — seq
   scan on a bounded table, idx_sync_queue_status_priority_created
   fits). The processor also dropped its fake-pagination loop, which
   re-fetched the same first batch forever once the backlog hit the
   batch size (no OFFSET in the query).
2. dbItemToSyncQueueItem dropped SyncData — even with a working
   enumerator, every flushed item arrived with an empty payload and
   failed JSON unmarshal. Dead code harboring silent breakage, exactly
   the plan's thesis.

Callers updated for the renamed RowLimit param (queue handler, offline
detector). TestH1_SyncQueueFlush_LandsAndPreservesAudio drives the real
flow: web PUT seeds audio 900s, a pending checkpoint-style queue row
applies percentage 0.8 through SaveProgress, audio survives, queue row
completes.

Also fixes dashboard_test.go's stale BuildSections signature (the
package's test binary has been unbuildable) — user-approved: tests
designed against old shapes are not worth keeping.
This commit is contained in:
John O'Keefe committed 2026-10-17 11:45:05 -04:00
1 parent d16ca4acbe
commit 6808c1babb
8 files changed
+101 -39

No files matched your search

+1 -1
View File
@@ -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 {
+17 -25
View File
@@ -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,