fix(store): include the cursor's own second in /changes deltas (BUG-2539)

items.updated_at / items.deleted_at are RFC3339 whole-second strings (store.now()), while the /changes cursor is a unix-millisecond value — normally the previous response's server_time. ItemsModifiedSince formatted that cursor with the same second precision, truncating it DOWN, then compared with a strict `>`. Every change landing in the cursor's own second compared equal and was dropped, permanently: the caller advances its cursor past that second and nothing reaches back. User-visible symptom: a bulk archive ~450ms after a page seeded its cursor left the item rendering as LIVE indefinitely — no banner, no redirect — while the server had deleted_at set. It was never archive-specific (updates were dropped identically); a missed update is usually re-delivered by the next event, a missed deletion never is. Compare inclusively against the truncated second instead. The boundary second may be re-delivered, which every consumer of this endpoint applies idempotently, and it is bounded to one second of changes per sync. Sub-second storage is the other fix and is a migration, not a one-liner: these comparisons are lexicographic on TEXT columns and mixing precisions inverts them ("…20.451Z" sorts BEFORE "…20Z"). Verified against a live instance with four cursors all strictly earlier than the archive in real time: two inside its second MISS, two in earlier seconds HIT. Tests: - TestItemsModifiedSince_SameSecondCursor — same-second leg plus a previous-second control. Fails 3/3 unfixed, passes 3/3 fixed; the control passes on both. - e2e bug-2539-sync-window — the banner must appear in the already-open page AND follow a /changes delta that carried the deletion, so a reload cannot satisfy it. The 450ms leg fails unfixed; 1200ms control passes on both. Claude-Session: https://claude.ai/code/session_01QGbUKZBAZoWdEgiTNWsXag
2026-09-11 13:28:57 +00:00 · 2026-08-13 14:31:57 +00:00
parent d8627968d0
commit 7905ed06d9
3 changed files with 222 additions and 7 deletions
@@ -4770,11 +4770,30 @@ func scanItems(rows *sql.Rows) ([]models.Item, error) {
 // items that were deleted (hard-deleted or archived) since the timestamp.
 //
 // The updated list includes both active AND recently archived items (those with
-// deleted_at > since). This lets the frontend update archived views correctly —
-// an item that was just archived needs its full data to appear in archived views,
-// not just its ID in the deleted list.
+// deleted_at at/after since). This lets the frontend update archived views
+// correctly — an item that was just archived needs its full data to appear in
+// archived views, not just its ID in the deleted list.
+//
+// The cursor comparison is INCLUSIVE of its own second, deliberately (BUG-2539).
+// items.updated_at / items.deleted_at are written at RFC3339 whole-second
+// precision (store.now()), while the caller's cursor is a unix-MILLISECOND value
+// — normally the previous response's server_time. Formatting that cursor for
+// comparison truncates it DOWN to the second, so with a strict `>` every change
+// that landed in the cursor's own second compares equal and is dropped. It is
+// dropped PERMANENTLY, because the caller then advances its cursor past that
+// second and no later query reaches back for it: a bulk archive ~450ms after a
+// page seeded its cursor left the item rendering as live indefinitely.
+//
+// `>=` against the truncated second may re-deliver changes from the boundary
+// second that the caller already has. That is the cheap direction to be wrong
+// in — every consumer of this endpoint applies server state idempotently — and
+// it is bounded to one second of changes per sync. Sub-second storage would be
+// the other fix, but items timestamps are second-precision throughout and
+// mixing formats in one column breaks the lexicographic ordering these string
+// comparisons rely on ("…20.451Z" sorts BEFORE "…20Z"), so precision is a
+// migration, not a one-line change.
 func (s *Store) ItemsModifiedSince(workspaceID string, since time.Time) (updated []models.Item, deletedIDs []string, err error) {
-	sinceStr := since.UTC().Format(time.RFC3339)
+	sinceStr := since.UTC().Truncate(time.Second).Format(time.RFC3339)

 	// Fetch updated items: active items modified since the timestamp,
 	// PLUS items archived since the timestamp (so archived views can update).
@@ -4791,8 +4810,8 @@ func (s *Store) ItemsModifiedSince(workspaceID string, since time.Time) (updated
 		LEFT JOIN users au ON au.id = i.assigned_user_id
 		LEFT JOIN agent_roles ar ON ar.id = i.agent_role_id
 		WHERE i.workspace_id = ?
-		  AND i.updated_at > ?
-		  AND (i.deleted_at IS NULL OR i.deleted_at > ?)
+		  AND i.updated_at >= ?
+		  AND (i.deleted_at IS NULL OR i.deleted_at >= ?)
 		ORDER BY i.updated_at ASC
 	`)

@@ -4811,7 +4830,7 @@ func (s *Store) ItemsModifiedSince(workspaceID string, since time.Time) (updated
 		SELECT id FROM items
 		WHERE workspace_id = ?
 		  AND deleted_at IS NOT NULL
-		  AND deleted_at > ?
+		  AND deleted_at >= ?
 	`)
 	delRows, err := s.db.Query(delQuery, workspaceID, sinceStr)
 	if err != nil {
@@ -0,0 +1,94 @@
+package store
+
+import (
+	"testing"
+	"time"
+
+	"github.com/PerpetualSoftware/pad/internal/models"
+)
+
+// BUG-2539. items.updated_at / items.deleted_at are RFC3339 whole-second
+// strings; the /changes cursor is a unix-millisecond value that gets formatted
+// with the same second precision, i.e. truncated DOWN. Under the original
+// strict `>` that made every change landing in the cursor's own second compare
+// equal and vanish — permanently, since the caller advances its cursor past
+// that second afterwards.
+//
+// Both cases below use a cursor that is strictly EARLIER than the mutation in
+// real time, so a correct implementation returns the change in BOTH. They
+// differ only in whether the cursor truncates to the same second as the write,
+// which is exactly the axis the bug lives on: the second leg is the control,
+// and it passed before the fix as well as after.
+func TestItemsModifiedSince_SameSecondCursor(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		// how far before the mutation the cursor sits
+		cursorLead time.Duration
+		// whether that lands the cursor inside the mutation's own second
+		sameSecond bool
+	}{
+		{name: "cursor inside the mutation's own second", cursorLead: 200 * time.Millisecond, sameSecond: true},
+		{name: "cursor in the previous second (control)", cursorLead: 1200 * time.Millisecond, sameSecond: false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			s := testStore(t)
+			ws := createTestWorkspace(t, s, "ChangesCursor")
+			coll := createTestCollection(t, s, ws.ID, "Tasks")
+
+			// Park the mutation ~500ms into a wall-clock second so a 200ms lead
+			// stays inside that second and a 1200ms lead cannot.
+			for {
+				if frac := time.Now().UnixMilli() % 1000; frac > 450 && frac < 550 {
+					break
+				}
+				time.Sleep(2 * time.Millisecond)
+			}
+
+			archived := createTestItem(t, s, ws.ID, coll.ID, "archived in-window", "")
+			updatedItem := createTestItem(t, s, ws.ID, coll.ID, "updated in-window", "")
+
+			mutateAt := time.Now()
+			cursor := mutateAt.Add(-tc.cursorLead)
+			if inSameSecond := cursor.UTC().Truncate(time.Second).Equal(mutateAt.UTC().Truncate(time.Second)); inSameSecond != tc.sameSecond {
+				t.Skipf("timing raced the second boundary (cursor same-second=%v, wanted %v)", inSameSecond, tc.sameSecond)
+			}
+
+			if err := s.DeleteItem(archived.ID); err != nil {
+				t.Fatalf("archive: %v", err)
+			}
+			newTitle := "updated in-window (touched)"
+			if _, err := s.UpdateItem(updatedItem.ID, models.ItemUpdate{Title: &newTitle}); err != nil {
+				t.Fatalf("update: %v", err)
+			}
+
+			updated, deletedIDs, err := s.ItemsModifiedSince(ws.ID, cursor)
+			if err != nil {
+				t.Fatalf("ItemsModifiedSince: %v", err)
+			}
+
+			var sawDeleted bool
+			for _, id := range deletedIDs {
+				if id == archived.ID {
+					sawDeleted = true
+				}
+			}
+			if !sawDeleted {
+				t.Errorf("archive at %s is missing from the deleted list for cursor %s (%v earlier, same-second=%v)",
+					mutateAt.UTC().Format(time.RFC3339Nano), cursor.UTC().Format(time.RFC3339Nano),
+					tc.cursorLead, tc.sameSecond)
+			}
+
+			var sawUpdated bool
+			for _, it := range updated {
+				if it.ID == updatedItem.ID {
+					sawUpdated = true
+				}
+			}
+			if !sawUpdated {
+				t.Errorf("update at %s is missing from the updated list for cursor %s (%v earlier, same-second=%v)",
+					mutateAt.UTC().Format(time.RFC3339Nano), cursor.UTC().Format(time.RFC3339Nano),
+					tc.cursorLead, tc.sameSecond)
+			}
+		})
+	}
+}
@@ -0,0 +1,102 @@
+import { expect } from '@playwright/test';
+import { test, quietCrossActorToasts } from './fixtures';
+
+/**
+ * BUG-2539 — a bulk archive landing in the same wall-clock second as the page's
+ * sync cursor was invisible to `/changes`, leaving the item rendered LIVE
+ * indefinitely (see internal/store/items.go::ItemsModifiedSince).
+ *
+ * The deterministic pin for the cursor arithmetic is the store-level
+ * `TestItemsModifiedSince_SameSecondCursor`. This spec covers the user-visible
+ * end of the same defect: the archived banner appears in the page that was
+ * already open, without a reload.
+ *
+ * Two legs. 450ms after navigation start is the offset that reproduced 7/8
+ * times before the fix (the archive lands just after the SSE connects and
+ * inside the cursor's own second); 1200ms is the control that always behaved.
+ * Both are deterministic AFTER the fix — the comparison no longer depends on
+ * where inside a second the mutation fell.
+ *
+ * Ground truth for archived-ness is the server's `deleted_at`, never the UI:
+ * asserting on the UI alone cannot distinguish "not archived" from "archived
+ * but not shown", which is the failure mode under test.
+ */
+
+// Offsets are measured from NAVIGATION START, not first paint — anchoring to
+// paint pushes the archive clear of the window and the bug does not reproduce.
+const DELAYS = [450, 1200];
+
+DELAYS.forEach((delay, idx) => {
+	test(`BUG-2539: bulk archive ${delay}ms after navigation start reaches the open item`, async ({
+		page,
+		context,
+		request,
+		fixture
+	}) => {
+		await quietCrossActorToasts(context);
+
+		// Own workspace: the shared suite workspace carries cross-spec mutation
+		// traffic that muddies a cursor-sensitive leg.
+		const slug = `b2539-${idx}-${Date.now().toString(36)}`;
+		const auth = { Authorization: `Bearer ${fixture.apiToken}` };
+		const wsResp = await request.post('/api/v1/workspaces', {
+			headers: auth,
+			data: { name: `BUG-2539 ${delay}`, slug, template: 'startup' }
+		});
+		expect(wsResp.ok(), await wsResp.text()).toBeTruthy();
+		const ws = (await wsResp.json()) as { slug: string };
+
+		const itemResp = await request.post(
+			`/api/v1/workspaces/${ws.slug}/collections/tasks/items`,
+			{ headers: auth, data: { title: `sync-window probe ${delay}` } }
+		);
+		expect(itemResp.ok(), await itemResp.text()).toBeTruthy();
+		const item = (await itemResp.json()) as { id: string; slug: string };
+
+		// Record whether the delta the page fetched actually carried the
+		// deletion. The banner assertion alone would also be satisfied by an
+		// unrelated mechanism (a full reload renders it too), so this is the
+		// leg that ties the banner to the sync path under test.
+		let deltaCarriedDeletion = false;
+		page.on('response', async (resp) => {
+			if (!resp.url().includes('/changes')) return;
+			try {
+				const body = (await resp.json()) as { deleted?: string[] };
+				if (body.deleted?.includes(item.id)) deltaCarriedDeletion = true;
+			} catch {
+				/* non-JSON body — ignore */
+			}
+		});
+
+		const itemUrl = `/${fixture.adminUsername}/${ws.slug}/tasks/${item.id}`;
+		// Fire the archive on a timer started at navigation start WITHOUT
+		// awaiting the page — awaiting first paint moves the archive out of the
+		// window under test.
+		const navStart = Date.now();
+		const archivePromise = (async () => {
+			await new Promise((r) => setTimeout(r, delay));
+			const resp = await request.post(`/api/v1/workspaces/${ws.slug}/items/bulk`, {
+				headers: auth,
+				data: { op: 'archive', ids: [item.id] }
+			});
+			return { ok: resp.ok(), body: await resp.text(), at: Date.now() - navStart };
+		})();
+		await page.goto(itemUrl, { waitUntil: 'domcontentloaded' });
+		const bulk = await archivePromise;
+		expect(bulk.ok, bulk.body).toBeTruthy();
+
+		// Premise: the archive really landed server-side.
+		const check = await request.get(`/api/v1/workspaces/${ws.slug}/items/${item.slug}`, {
+			headers: auth
+		});
+		const checked = (await check.json()) as { deleted_at: string | null };
+		expect(checked.deleted_at, 'archive must land server-side').not.toBeNull();
+
+		// The open page must learn about it without a reload.
+		await expect(page.locator('.archived-banner')).toBeVisible({ timeout: 10_000 });
+		expect(
+			deltaCarriedDeletion,
+			'the banner must follow a /changes delta carrying the deletion, not an unrelated reload'
+		).toBeTruthy();
+	});
+});