| 1 | package memoria |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "net/http" |
| 8 | "net/http/httptest" |
| 9 | "strings" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "amadan.net/oficina/suite" |
| 14 | |
| 15 | "amadan.net/oficina/memoria/internal/memoria/sibling" |
| 16 | ) |
| 17 | |
| 18 | // feedServer is a sibling that serves a scripted sequence of S1 pages and |
| 19 | // an S3 body per item. |
| 20 | type feedServer struct { |
| 21 | t *testing.T |
| 22 | // pages is consulted by cursor: the empty string is the first page. |
| 23 | pages map[string]suite.Feed |
| 24 | bodies map[string]string |
| 25 | failS1 map[string]bool |
| 26 | failS3 map[string]bool |
| 27 | // brokeS3 is a 5xx rather than a 404: an ERROR, where failS3 is a |
| 28 | // refusal. The two must not be confused — a 404 is counted and moves |
| 29 | // on, a 500 stops the page — and no fixture distinguished them until |
| 30 | // a mutation showed the cursor could advance mid-page unnoticed. |
| 31 | brokeS3 map[string]bool |
| 32 | // forbidS3 answers 403, which is a CONTRACT BREACH and a third thing |
| 33 | // again: the sibling is telling callers which ids exist. |
| 34 | forbidS3 map[string]bool |
| 35 | s1Calls int |
| 36 | s3Calls int |
| 37 | } |
| 38 | |
| 39 | func (f *feedServer) start(t *testing.T) (sibling.Sibling, *sibling.Client) { |
| 40 | t.Helper() |
| 41 | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 42 | switch { |
| 43 | case r.URL.Path == suite.RouteS1: |
| 44 | f.s1Calls++ |
| 45 | cur := r.URL.Query().Get("cursor") |
| 46 | if f.failS1[cur] { |
| 47 | w.WriteHeader(http.StatusInternalServerError) |
| 48 | // A well-formed empty feed with a bad status: the shape |
| 49 | // that would authorise a purge if the status were ignored. |
| 50 | _, _ = w.Write([]byte(`{"cursor":"","rows":[]}`)) |
| 51 | return |
| 52 | } |
| 53 | page, ok := f.pages[cur] |
| 54 | if !ok { |
| 55 | f.t.Errorf("sibling asked for an unscripted cursor %q", cur) |
| 56 | w.WriteHeader(http.StatusInternalServerError) |
| 57 | return |
| 58 | } |
| 59 | _ = json.NewEncoder(w).Encode(page) |
| 60 | case strings.HasPrefix(r.URL.Path, suite.RouteS3+"/"): |
| 61 | f.s3Calls++ |
| 62 | id := strings.TrimPrefix(r.URL.Path, suite.RouteS3+"/") |
| 63 | if f.forbidS3[id] { |
| 64 | w.WriteHeader(http.StatusForbidden) |
| 65 | return |
| 66 | } |
| 67 | if f.brokeS3[id] { |
| 68 | w.WriteHeader(http.StatusInternalServerError) |
| 69 | return |
| 70 | } |
| 71 | if f.failS3[id] { |
| 72 | w.WriteHeader(http.StatusNotFound) |
| 73 | return |
| 74 | } |
| 75 | body, ok := f.bodies[id] |
| 76 | if !ok { |
| 77 | w.WriteHeader(http.StatusNotFound) |
| 78 | return |
| 79 | } |
| 80 | _, _ = w.Write([]byte(body)) |
| 81 | default: |
| 82 | f.t.Errorf("sibling asked for %s %s", r.Method, r.URL.Path) |
| 83 | w.WriteHeader(http.StatusNotFound) |
| 84 | } |
| 85 | })) |
| 86 | t.Cleanup(srv.Close) |
| 87 | s := sibling.Sibling{Name: "Docs", BaseURL: srv.URL, Token: "consumer-token"} |
| 88 | return s, sibling.NewForTest() |
| 89 | } |
| 90 | |
| 91 | func row(id string, rev int64, gone bool) suite.FeedRow { |
| 92 | return suite.FeedRow{ItemID: id, Revision: rev, Gone: gone, TouchedAt: time.Unix(0, 0).UTC()} |
| 93 | } |
| 94 | |
| 95 | // Gate 7 where it actually bites: **a failed poll advances nothing and |
| 96 | // deletes nothing.** |
| 97 | // |
| 98 | // The clients already refuse to turn a failure into an empty feed. This is |
| 99 | // the layer above, where the consequences live: a cursor moved past a page |
| 100 | // nobody applied skips changes the sibling will never offer again, and it |
| 101 | // does so silently, because the sibling has no way to know its consumer |
| 102 | // stopped listening. |
| 103 | // |
| 104 | // TWO-SIDED, and the first subtest is what makes the rest mean anything: a |
| 105 | // Poll that never stores would satisfy every "nothing was deleted" |
| 106 | // assertion below. |
| 107 | // |
| 108 | // MUTATION-TESTED 2026-09-07, each verified red then reverted: |
| 109 | // |
| 110 | // setCursor called before the rows are applied |
| 111 | // -> the mid-page-failure case red (the cursor moved) |
| 112 | // the S1 error swallowed and the loop continued |
| 113 | // -> the failed-poll case red |
| 114 | // a `gone` row applied by deleting on a failed S3 instead |
| 115 | // -> the missing case red (the document went) |
| 116 | // Missing counted as Deleted |
| 117 | // -> the missing case red |
| 118 | // the empty final page treated as More |
| 119 | // -> the completion case red |
| 120 | func TestAFailedPollAdvancesNothing(t *testing.T) { |
| 121 | ctx := context.Background() |
| 122 | |
| 123 | t.Run("a good poll stores the sibling's documents", func(t *testing.T) { |
| 124 | f := &feedServer{ |
| 125 | t: t, |
| 126 | pages: map[string]suite.Feed{ |
| 127 | // Revisions 7 and 9, not 1: the sibling issues these and |
| 128 | // memoria stores what it was given. With a fixture at |
| 129 | // revision 1, hard-coding `SiblingRevision: 1` passes — |
| 130 | // a mutation did. |
| 131 | "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 7, false), row("b", 9, false)}}, |
| 132 | "c1": {Cursor: "c1", Rows: []suite.FeedRow{}}, |
| 133 | }, |
| 134 | bodies: map[string]string{ |
| 135 | "a": "The quarterly planning process runs in March.", |
| 136 | "b": "Expenses are filed monthly.", |
| 137 | }, |
| 138 | } |
| 139 | sib, client := f.start(t) |
| 140 | s := storeForTest(t) |
| 141 | |
| 142 | res, err := s.Poll(ctx, client, sib) |
| 143 | if err != nil { |
| 144 | t.Fatalf("poll: %v", err) |
| 145 | } |
| 146 | if res.Stored != 2 || res.Deleted != 0 || res.Missing != 0 || res.More { |
| 147 | t.Errorf("result = %+v, want 2 stored and nothing else", res) |
| 148 | } |
| 149 | // Stored WITH their coordinates, or the ruling has nothing to ask |
| 150 | // about and every one of them withholds forever. |
| 151 | assertHeld(t, s, "Docs", "a", 7) |
| 152 | assertHeld(t, s, "Docs", "b", 9) |
| 153 | }) |
| 154 | |
| 155 | t.Run("a row failing mid-page leaves the cursor before that page", func(t *testing.T) { |
| 156 | // **This is the file's central claim, and nothing reached it.** |
| 157 | // The other failure cases fail at the S1 fetch, where no row has |
| 158 | // been applied yet and storing the cursor first is indisitinguishable |
| 159 | // from storing it last. A mutation that moved setCursor above the |
| 160 | // apply loop passed every one of them. |
| 161 | // |
| 162 | // Here the page is fetched fine and a row inside it fails. With |
| 163 | // the cursor stored first, page one is marked done while document |
| 164 | // "b" was never stored — and the sibling will not mention it |
| 165 | // again, because from its side the consumer read that page. |
| 166 | f := &feedServer{ |
| 167 | t: t, |
| 168 | pages: map[string]suite.Feed{ |
| 169 | "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 1, false), row("b", 1, false)}}, |
| 170 | "c1": {Cursor: "c1", Rows: []suite.FeedRow{}}, |
| 171 | }, |
| 172 | bodies: map[string]string{"a": "Applied.", "b": "Never fetched."}, |
| 173 | brokeS3: map[string]bool{"b": true}, |
| 174 | } |
| 175 | sib, client := f.start(t) |
| 176 | s := storeForTest(t) |
| 177 | |
| 178 | if _, err := s.Poll(ctx, client, sib); err == nil { |
| 179 | t.Fatal("a 500 from S3 mid-page came back as a completed poll") |
| 180 | } |
| 181 | if got := cursorOf(t, s, "Docs"); got != "" { |
| 182 | t.Errorf("cursor = %q, want empty: a page with an unapplied row in "+ |
| 183 | "it must not be marked read, or the sibling never offers it "+ |
| 184 | "again and the document is lost silently", got) |
| 185 | } |
| 186 | // Row "a" was applied before the failure and stands — re-applying |
| 187 | // it next time is the cheap failure, and the point of the |
| 188 | // assertion is that the page will be re-offered at all. |
| 189 | assertHeld(t, s, "Docs", "a", 1) |
| 190 | if heldCount(t, s, "Docs") != 1 { |
| 191 | t.Errorf("%d documents held; the row that failed must not be "+ |
| 192 | "stored", heldCount(t, s, "Docs")) |
| 193 | } |
| 194 | |
| 195 | // Let it work, and the whole page arrives. |
| 196 | f.brokeS3 = nil |
| 197 | if _, err := s.Poll(ctx, client, sib); err != nil { |
| 198 | t.Fatalf("second poll: %v", err) |
| 199 | } |
| 200 | assertHeld(t, s, "Docs", "b", 1) |
| 201 | }) |
| 202 | |
| 203 | t.Run("a page carrying a row that names nothing is refused whole", func(t *testing.T) { |
| 204 | // A row with no item id can neither be stored nor deleted, and |
| 205 | // passing over it would advance the cursor past a change memoria |
| 206 | // never applied. |
| 207 | // |
| 208 | // **The refusal is the MODULE's**, and this test says so on |
| 209 | // purpose. memoria had its own check; it could not fail, because |
| 210 | // DecodeS1 rejects the page before applyRow sees a row — a |
| 211 | // mutation deleting memoria's guard left the suite green. So the |
| 212 | // guard is gone and this asserts the thing memoria actually |
| 213 | // depends on. It goes red if the module ever softens, which is |
| 214 | // the day memoria owes a check of its own. |
| 215 | f := &feedServer{ |
| 216 | t: t, |
| 217 | pages: map[string]suite.Feed{ |
| 218 | "": {Cursor: "c1", Rows: []suite.FeedRow{row("", 1, false)}}, |
| 219 | // A clean second page, so SKIPPING the malformed row would |
| 220 | // produce a completely successful poll. Without it the |
| 221 | // poll fails anyway — on an unscripted cursor — and the |
| 222 | // assertion passes for the wrong reason. |
| 223 | "c1": {Cursor: "c1", Rows: []suite.FeedRow{}}, |
| 224 | }, |
| 225 | bodies: map[string]string{}, |
| 226 | } |
| 227 | sib, client := f.start(t) |
| 228 | s := storeForTest(t) |
| 229 | |
| 230 | if _, err := s.Poll(ctx, client, sib); err == nil { |
| 231 | t.Fatal("a feed row with no item id was passed over silently") |
| 232 | } |
| 233 | if got := cursorOf(t, s, "Docs"); got != "" { |
| 234 | t.Errorf("cursor = %q, want empty: a page memoria could not apply "+ |
| 235 | "whole must not be marked read", got) |
| 236 | } |
| 237 | }) |
| 238 | |
| 239 | t.Run("a failed page leaves the cursor where it was", func(t *testing.T) { |
| 240 | f := &feedServer{ |
| 241 | t: t, |
| 242 | pages: map[string]suite.Feed{ |
| 243 | "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 1, false)}}, |
| 244 | "c1": {Cursor: "c2", Rows: []suite.FeedRow{row("b", 1, false)}}, |
| 245 | }, |
| 246 | bodies: map[string]string{"a": "First document.", "b": "Second document."}, |
| 247 | failS1: map[string]bool{"c1": true}, |
| 248 | } |
| 249 | sib, client := f.start(t) |
| 250 | s := storeForTest(t) |
| 251 | |
| 252 | if _, err := s.Poll(ctx, client, sib); err == nil { |
| 253 | t.Fatal("a 500 from S1 came back as a completed poll") |
| 254 | } |
| 255 | // Page one was applied and stands. |
| 256 | assertHeld(t, s, "Docs", "a", 1) |
| 257 | // And the position is page one's, so page two is offered again |
| 258 | // rather than skipped forever. |
| 259 | if got := cursorOf(t, s, "Docs"); got != "c1" { |
| 260 | t.Errorf("cursor = %q, want c1: a poll that failed on the second "+ |
| 261 | "page must leave the position at the first", got) |
| 262 | } |
| 263 | |
| 264 | // Now let it succeed, and the page it could not read arrives. |
| 265 | f.failS1 = nil |
| 266 | f.pages["c2"] = suite.Feed{Cursor: "c2", Rows: []suite.FeedRow{}} |
| 267 | if _, err := s.Poll(ctx, client, sib); err != nil { |
| 268 | t.Fatalf("second poll: %v", err) |
| 269 | } |
| 270 | assertHeld(t, s, "Docs", "b", 1) |
| 271 | }) |
| 272 | |
| 273 | t.Run("a listed item the sibling will not serve is counted, not deleted", func(t *testing.T) { |
| 274 | f := &feedServer{ |
| 275 | t: t, |
| 276 | pages: map[string]suite.Feed{ |
| 277 | "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 1, false)}}, |
| 278 | "c1": {Cursor: "c1", Rows: []suite.FeedRow{}}, |
| 279 | }, |
| 280 | bodies: map[string]string{"a": "Held already."}, |
| 281 | } |
| 282 | sib, client := f.start(t) |
| 283 | s := storeForTest(t) |
| 284 | if _, err := s.Poll(ctx, client, sib); err != nil { |
| 285 | t.Fatalf("first poll: %v", err) |
| 286 | } |
| 287 | assertHeld(t, s, "Docs", "a", 1) |
| 288 | |
| 289 | // The sibling now lists it at a new revision and refuses to serve |
| 290 | // it. §4.2 makes that 404 cover both "gone" and "not |
| 291 | // team-visible", so memoria cannot tell which — and deleting on it |
| 292 | // would let a visibility change erase history the sibling still |
| 293 | // has. |
| 294 | f.pages[""] = suite.Feed{Cursor: "c2", Rows: []suite.FeedRow{row("a", 2, false)}} |
| 295 | f.pages["c2"] = suite.Feed{Cursor: "c2", Rows: []suite.FeedRow{}} |
| 296 | f.failS3 = map[string]bool{"a": true} |
| 297 | |
| 298 | // The store under test is already at cursor "c1", so it is page |
| 299 | // "c1" that must now carry the re-announced row — not page "". |
| 300 | // Both are set: the first line above is what a sibling starting |
| 301 | // fresh would serve, and this is what ours will actually ask for. |
| 302 | f.pages["c1"] = suite.Feed{Cursor: "c2", Rows: []suite.FeedRow{row("a", 2, false)}} |
| 303 | res, err := s.Poll(ctx, client, sib) |
| 304 | if err != nil { |
| 305 | t.Fatalf("second poll: %v", err) |
| 306 | } |
| 307 | if res.Missing != 1 { |
| 308 | t.Errorf("result = %+v, want 1 missing", res) |
| 309 | } |
| 310 | if res.Deleted != 0 { |
| 311 | t.Errorf("a 404 on S3 deleted %d documents: an item that is not "+ |
| 312 | "team-visible is indistinguishable from one that is gone, and "+ |
| 313 | "acting on that erases history the sibling still has", res.Deleted) |
| 314 | } |
| 315 | // Still held, at the revision memoria could actually fetch. |
| 316 | assertHeld(t, s, "Docs", "a", 1) |
| 317 | }) |
| 318 | |
| 319 | t.Run("a gone row deletes, so the case above is not vacuous", func(t *testing.T) { |
| 320 | f := &feedServer{ |
| 321 | t: t, |
| 322 | pages: map[string]suite.Feed{ |
| 323 | "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 1, false)}}, |
| 324 | "c1": {Cursor: "c2", Rows: []suite.FeedRow{row("a", 2, true)}}, |
| 325 | "c2": {Cursor: "c2", Rows: []suite.FeedRow{}}, |
| 326 | }, |
| 327 | bodies: map[string]string{"a": "Here for now."}, |
| 328 | } |
| 329 | sib, client := f.start(t) |
| 330 | s := storeForTest(t) |
| 331 | if _, err := s.Poll(ctx, client, sib); err != nil { |
| 332 | t.Fatalf("poll: %v", err) |
| 333 | } |
| 334 | // One poll walks both pages: stored, then tombstoned. |
| 335 | if held := heldCount(t, s, "Docs"); held != 0 { |
| 336 | t.Errorf("%d documents survived a tombstone", held) |
| 337 | } |
| 338 | }) |
| 339 | } |
| 340 | |
| 341 | // A sibling whose cursor never advances must not loop forever, and the |
| 342 | // poll must say it did not finish. |
| 343 | // |
| 344 | // "Stopped looking" and "there is no more" are the same value to a caller |
| 345 | // that only checks the error, which is why More exists. |
| 346 | // |
| 347 | // MUTATION-TESTED 2026-09-07: drop the maxFeedPages bound and the test |
| 348 | // hangs rather than failing — which is itself the finding, so the bound is |
| 349 | // asserted by the call count as well. Reverted. |
| 350 | func TestAFeedThatNeverAdvancesStopsAndSaysSo(t *testing.T) { |
| 351 | f := &feedServer{ |
| 352 | t: t, |
| 353 | // A page that hands back the cursor it was asked with. The module |
| 354 | // will not serve an empty cursor at all, so this — not a blank — |
| 355 | // is what "the feed did not advance" looks like on the wire. |
| 356 | pages: map[string]suite.Feed{ |
| 357 | "": {Cursor: "same", Rows: []suite.FeedRow{row("a", 1, false)}}, |
| 358 | "same": {Cursor: "same", Rows: []suite.FeedRow{row("a", 1, false)}}, |
| 359 | }, |
| 360 | bodies: map[string]string{"a": "Going nowhere."}, |
| 361 | } |
| 362 | sib, client := f.start(t) |
| 363 | s := storeForTest(t) |
| 364 | |
| 365 | res, err := s.Poll(context.Background(), client, sib) |
| 366 | if err != nil { |
| 367 | t.Fatalf("poll: %v", err) |
| 368 | } |
| 369 | if !res.More { |
| 370 | t.Error("a feed that never advanced its cursor reported a complete " + |
| 371 | "poll; the corpus would be called current") |
| 372 | } |
| 373 | // Two: the first page advanced the cursor, the second handed back the |
| 374 | // same one and the poll stopped. Bounded, not walked to maxFeedPages. |
| 375 | if f.s1Calls != 2 { |
| 376 | t.Errorf("%d S1 calls for a feed that stops advancing, want 2: a feed "+ |
| 377 | "repeating itself must not be walked to the page bound", f.s1Calls) |
| 378 | } |
| 379 | } |
| 380 | |
| 381 | func assertHeld(t *testing.T, s *Store, app, itemID string, revision int64) { |
| 382 | t.Helper() |
| 383 | external, err := s.externalID(itemID) |
| 384 | if err != nil { |
| 385 | t.Fatalf("externalID: %v", err) |
| 386 | } |
| 387 | var d Document |
| 388 | if err := s.db.G.Where("source = ? AND external_id = ?", app, external).First(&d).Error; err != nil { |
| 389 | t.Fatalf("%s/%s not held: %v", app, itemID, err) |
| 390 | } |
| 391 | if d.SiblingRevision != revision { |
| 392 | t.Errorf("%s/%s held at revision %d, want %d", app, itemID, d.SiblingRevision, revision) |
| 393 | } |
| 394 | if len(d.SiblingItemID) == 0 { |
| 395 | t.Errorf("%s/%s stored without its coordinate, so nothing can be "+ |
| 396 | "asked about it and it withholds forever", app, itemID) |
| 397 | } |
| 398 | } |
| 399 | |
| 400 | func heldCount(t *testing.T, s *Store, app string) int { |
| 401 | t.Helper() |
| 402 | var n int64 |
| 403 | if err := s.db.G.Model(&Document{}).Where("source = ?", app).Count(&n).Error; err != nil { |
| 404 | t.Fatalf("count: %v", err) |
| 405 | } |
| 406 | return int(n) |
| 407 | } |
| 408 | |
| 409 | func cursorOf(t *testing.T, s *Store, app string) string { |
| 410 | t.Helper() |
| 411 | got, err := s.cursor(context.Background(), app) |
| 412 | if err != nil { |
| 413 | t.Fatalf("cursor: %v", err) |
| 414 | } |
| 415 | return got |
| 416 | } |
| 417 | |
| 418 | // The cursor is sealed at rest, and a cursor that will not open is an |
| 419 | // error rather than an empty one. |
| 420 | // |
| 421 | // An empty cursor is not a degraded value: it is the instruction "start |
| 422 | // from the beginning", which would re-announce the sibling's whole corpus. |
| 423 | // Reading a corrupted row as that instruction is the plausible-empty-list |
| 424 | // failure wearing its other face — a confident restart rather than a |
| 425 | // confident absence. |
| 426 | // |
| 427 | // MUTATION-TESTED 2026-09-07: return ("", nil) when Open fails, and the |
| 428 | // second half goes red. Reverted. |
| 429 | func TestACursorIsSealedAndACorruptOneIsAnError(t *testing.T) { |
| 430 | ctx := context.Background() |
| 431 | s := storeForTest(t) |
| 432 | if err := s.setCursor(ctx, "Docs", "opaque-cursor-value"); err != nil { |
| 433 | t.Fatalf("setCursor: %v", err) |
| 434 | } |
| 435 | |
| 436 | var stored SiblingFeed |
| 437 | if err := s.db.G.Where("app = ?", "Docs").First(&stored).Error; err != nil { |
| 438 | t.Fatalf("read back: %v", err) |
| 439 | } |
| 440 | // The presence assertion: it round-trips, so the absence check below |
| 441 | // is about sealing rather than about an empty column. |
| 442 | if got := cursorOf(t, s, "Docs"); got != "opaque-cursor-value" { |
| 443 | t.Fatalf("cursor round-tripped as %q", got) |
| 444 | } |
| 445 | if strings.Contains(string(stored.SealedCursor), "opaque-cursor-value") { |
| 446 | t.Error("the cursor is at rest in plaintext") |
| 447 | } |
| 448 | |
| 449 | // Corrupt it. The row is present and unreadable, which must not read |
| 450 | // as "never polled". |
| 451 | if err := s.db.G.Model(&SiblingFeed{}).Where("id = ?", stored.ID). |
| 452 | Update("sealed_cursor", []byte("not a sealed value at all")).Error; err != nil { |
| 453 | t.Fatalf("corrupt: %v", err) |
| 454 | } |
| 455 | if got, err := s.cursor(ctx, "Docs"); err == nil { |
| 456 | t.Errorf("an unreadable cursor came back as %q with no error: that is "+ |
| 457 | "the instruction to re-announce the sibling's entire corpus", got) |
| 458 | } |
| 459 | } |
| 460 | |
| 461 | // PolledAt says when the corpus was last known current, so a poll that |
| 462 | // failed must not move it. §8.1: a failed or stale S1 makes the page state |
| 463 | // what it is current as of, rather than implying freshness. |
| 464 | // |
| 465 | // MUTATION-TESTED 2026-09-07: set polled_at before walking the pages and |
| 466 | // this goes red. Reverted. |
| 467 | func TestAFailedPollDoesNotClaimFreshness(t *testing.T) { |
| 468 | f := &feedServer{ |
| 469 | t: t, |
| 470 | pages: map[string]suite.Feed{"": {Cursor: "c1", Rows: []suite.FeedRow{}}}, |
| 471 | failS1: map[string]bool{"": true}, |
| 472 | } |
| 473 | sib, client := f.start(t) |
| 474 | s := storeForTest(t) |
| 475 | |
| 476 | if _, err := s.Poll(context.Background(), client, sib); err == nil { |
| 477 | t.Fatal("a failing S1 reported a completed poll") |
| 478 | } |
| 479 | var row SiblingFeed |
| 480 | err := s.db.G.Where("app = ?", "Docs").First(&row).Error |
| 481 | if err == nil && !row.PolledAt.IsZero() { |
| 482 | t.Errorf("a poll that never read a page recorded PolledAt=%v; the "+ |
| 483 | "sources page would report the corpus current as of a moment "+ |
| 484 | "nothing was read", row.PolledAt) |
| 485 | } |
| 486 | if err != nil && !strings.Contains(fmt.Sprint(err), "record not found") { |
| 487 | t.Fatalf("read back: %v", err) |
| 488 | } |
| 489 | } |
| 490 | |
| 491 | // **An empty item is skipped, and the feed keeps moving.** |
| 492 | // |
| 493 | // This was the opposite: `AllowTitleOnly` was passed with no Title, so |
| 494 | // Ingest refused with "neither a body nor a title", the row failed, the |
| 495 | // cursor stayed, and every subsequent poll of that sibling failed on the |
| 496 | // same row. One stub page in Docs and nothing from Docs was ever ingested |
| 497 | // again — a permanent, silent halt with only "sibling poll failed" to show |
| 498 | // for it. A review found it by putting a stub first in a page. |
| 499 | // |
| 500 | // TWO-SIDED: the real document after the stub must arrive, or "refuse |
| 501 | // everything" satisfies the count. |
| 502 | // |
| 503 | // MUTATION-TESTED 2026-09-07, each verified red then reverted: |
| 504 | // |
| 505 | // restore AllowTitleOnly and drop the empty check |
| 506 | // -> red: the poll errors and the real document is not held |
| 507 | // skip the empty row without counting it |
| 508 | // -> red on Empty |
| 509 | // treat whitespace as a body |
| 510 | // -> red: Ingest refuses " " the same way |
| 511 | func TestAnEmptyItemIsSkippedNotWedged(t *testing.T) { |
| 512 | ctx := context.Background() |
| 513 | f := &feedServer{ |
| 514 | t: t, |
| 515 | pages: map[string]suite.Feed{ |
| 516 | "": {Cursor: "c1", Rows: []suite.FeedRow{ |
| 517 | row("stub", 1, false), |
| 518 | row("blank", 1, false), |
| 519 | row("real", 1, false), |
| 520 | }}, |
| 521 | "c1": {Cursor: "c1", Rows: []suite.FeedRow{}}, |
| 522 | }, |
| 523 | bodies: map[string]string{ |
| 524 | "stub": "", |
| 525 | "blank": " \n\t ", |
| 526 | "real": "The quarterly planning process runs in March.", |
| 527 | }, |
| 528 | } |
| 529 | sib, client := f.start(t) |
| 530 | s := storeForTest(t) |
| 531 | |
| 532 | res, err := s.Poll(ctx, client, sib) |
| 533 | if err != nil { |
| 534 | t.Fatalf("a stub page stopped the poll: %v", err) |
| 535 | } |
| 536 | if res.Empty != 2 { |
| 537 | t.Errorf("result = %+v, want 2 empty", res) |
| 538 | } |
| 539 | if res.Stored != 1 { |
| 540 | t.Errorf("result = %+v, want 1 stored", res) |
| 541 | } |
| 542 | // The document after the stubs arrived, which is the whole point. |
| 543 | assertHeld(t, s, "Docs", "real", 1) |
| 544 | if heldCount(t, s, "Docs") != 1 { |
| 545 | t.Errorf("%d documents held; an empty item has no passages to "+ |
| 546 | "retrieve and must not become one", heldCount(t, s, "Docs")) |
| 547 | } |
| 548 | // And the feed moved on, so the next poll does not re-read this page. |
| 549 | if got := cursorOf(t, s, "Docs"); got != "c1" { |
| 550 | t.Errorf("cursor = %q, want c1", got) |
| 551 | } |
| 552 | } |
| 553 | |
| 554 | // **A revision bump with an identical body must still be recorded.** |
| 555 | // |
| 556 | // The two revisions answer different questions. memoria's counts its own |
| 557 | // edits; the sibling's is the token S2 is asked about, and a sibling |
| 558 | // issues a new one for reasons memoria cannot see — a permissions change, |
| 559 | // a metadata edit, a re-save that touched no bytes. |
| 560 | // |
| 561 | // Ingest's unchanged branch returned without touching it, and the failure |
| 562 | // was permanent and invisible: the sibling announces revision 8 of a body |
| 563 | // identical to 7, memoria keeps 7, and every query thereafter asks S2 |
| 564 | // about `(item, 7)`. The sibling answers STALE — correctly, it has moved |
| 565 | // on — so the passage is withheld loudly on every page forever, and |
| 566 | // re-polling cannot fix it because the body never changes. Found by a |
| 567 | // review, 2026-09-07. |
| 568 | // |
| 569 | // MUTATION-TESTED 2026-09-07, each verified red then reverted: |
| 570 | // |
| 571 | // restore the bare `unchanged = true; return nil` |
| 572 | // -> red, the row stays at 7 |
| 573 | // update sibling_revision but also bump Revision |
| 574 | // -> red: memoria's own counter must not move for a body that did |
| 575 | // not, or §5.2's provenance invalidates artefacts for nothing |
| 576 | func TestARevisionBumpOnAnIdenticalBodyIsRecorded(t *testing.T) { |
| 577 | ctx := context.Background() |
| 578 | const body = "The quarterly planning process runs in March." |
| 579 | f := &feedServer{ |
| 580 | t: t, |
| 581 | pages: map[string]suite.Feed{ |
| 582 | "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 7, false)}}, |
| 583 | "c1": {Cursor: "c1", Rows: []suite.FeedRow{}}, |
| 584 | }, |
| 585 | bodies: map[string]string{"a": body}, |
| 586 | } |
| 587 | sib, client := f.start(t) |
| 588 | s := storeForTest(t) |
| 589 | if _, err := s.Poll(ctx, client, sib); err != nil { |
| 590 | t.Fatalf("first poll: %v", err) |
| 591 | } |
| 592 | assertHeld(t, s, "Docs", "a", 7) |
| 593 | before := documentOf(t, s, "Docs", "a") |
| 594 | |
| 595 | // The sibling moves to 8. Same bytes. |
| 596 | f.pages["c1"] = suite.Feed{Cursor: "c2", Rows: []suite.FeedRow{row("a", 8, false)}} |
| 597 | f.pages["c2"] = suite.Feed{Cursor: "c2", Rows: []suite.FeedRow{}} |
| 598 | if _, err := s.Poll(ctx, client, sib); err != nil { |
| 599 | t.Fatalf("second poll: %v", err) |
| 600 | } |
| 601 | |
| 602 | after := documentOf(t, s, "Docs", "a") |
| 603 | if after.SiblingRevision != 8 { |
| 604 | t.Errorf("held at sibling revision %d after the sibling announced 8: "+ |
| 605 | "S2 is asked about %d, the sibling answers STALE, and the passage "+ |
| 606 | "is withheld on every query forever — re-polling cannot fix it "+ |
| 607 | "because the body never changes", |
| 608 | after.SiblingRevision, after.SiblingRevision) |
| 609 | } |
| 610 | if after.Revision != before.Revision { |
| 611 | t.Errorf("memoria's own revision moved from %d to %d for a body that "+ |
| 612 | "did not change; §5.2 would invalidate every artefact built from "+ |
| 613 | "it for nothing", before.Revision, after.Revision) |
| 614 | } |
| 615 | } |
| 616 | |
| 617 | // PolledAt is a claim about COMPLETENESS, and only a drained feed may make |
| 618 | // it. |
| 619 | // |
| 620 | // A position is "how far I got"; a timestamp is "and there was nothing |
| 621 | // after it". These were written together, once per page, so a multi-page |
| 622 | // catch-up that failed on page four — or stopped at the page bound with |
| 623 | // more to come — left PolledAt at the moment of the last page it managed. |
| 624 | // The sources page would then report the corpus current as of a time it |
| 625 | // demonstrably was not. §8.1: *a failed or stale S1 states what the corpus |
| 626 | // is current as of, rather than implying freshness.* |
| 627 | // |
| 628 | // **The presence half is the one that was missing.** The existing test |
| 629 | // only ever failed on page one, so "PolledAt was not set" passed against a |
| 630 | // Poll that never set it at all — and a mutation deleting the write |
| 631 | // entirely stayed green. This asserts a good poll DOES set it first. |
| 632 | // |
| 633 | // MUTATION-TESTED 2026-09-07, each verified red then reverted: |
| 634 | // |
| 635 | // setCursor writes polled_at again |
| 636 | // -> the second-page failure case red, and the More case red |
| 637 | // the drained branch stops writing polled_at |
| 638 | // -> the presence case red |
| 639 | func TestPolledAtMeansTheFeedWasDrained(t *testing.T) { |
| 640 | ctx := context.Background() |
| 641 | |
| 642 | t.Run("a completed poll records when", func(t *testing.T) { |
| 643 | f := &feedServer{ |
| 644 | t: t, |
| 645 | pages: map[string]suite.Feed{ |
| 646 | "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 1, false)}}, |
| 647 | "c1": {Cursor: "c1", Rows: []suite.FeedRow{}}, |
| 648 | }, |
| 649 | bodies: map[string]string{"a": "Held."}, |
| 650 | } |
| 651 | sib, client := f.start(t) |
| 652 | s := storeForTest(t) |
| 653 | if _, err := s.Poll(ctx, client, sib); err != nil { |
| 654 | t.Fatalf("poll: %v", err) |
| 655 | } |
| 656 | if polledAt(t, s, "Docs").IsZero() { |
| 657 | t.Error("a poll that drained the feed recorded no time; the sources " + |
| 658 | "page has nothing to say the corpus is current as of") |
| 659 | } |
| 660 | }) |
| 661 | |
| 662 | t.Run("a poll failing on a later page claims nothing", func(t *testing.T) { |
| 663 | f := &feedServer{ |
| 664 | t: t, |
| 665 | pages: map[string]suite.Feed{ |
| 666 | "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 1, false)}}, |
| 667 | "c1": {Cursor: "c2", Rows: []suite.FeedRow{row("b", 1, false)}}, |
| 668 | }, |
| 669 | bodies: map[string]string{"a": "First.", "b": "Second."}, |
| 670 | failS1: map[string]bool{"c1": true}, |
| 671 | } |
| 672 | sib, client := f.start(t) |
| 673 | s := storeForTest(t) |
| 674 | if _, err := s.Poll(ctx, client, sib); err == nil { |
| 675 | t.Fatal("a 500 on page two came back as a completed poll") |
| 676 | } |
| 677 | // Page one WAS applied and the cursor moved — so the row exists, |
| 678 | // and this is an assertion about the column rather than about the |
| 679 | // row's absence. |
| 680 | if got := cursorOf(t, s, "Docs"); got != "c1" { |
| 681 | t.Fatalf("cursor = %q, want c1", got) |
| 682 | } |
| 683 | if at := polledAt(t, s, "Docs"); !at.IsZero() { |
| 684 | t.Errorf("a poll that failed on page two recorded PolledAt=%v: the "+ |
| 685 | "sources page would call the corpus current as of a moment "+ |
| 686 | "there was demonstrably more to read", at) |
| 687 | } |
| 688 | }) |
| 689 | |
| 690 | t.Run("a poll that stopped short claims nothing", func(t *testing.T) { |
| 691 | f := &feedServer{ |
| 692 | t: t, |
| 693 | pages: map[string]suite.Feed{ |
| 694 | "": {Cursor: "same", Rows: []suite.FeedRow{row("a", 1, false)}}, |
| 695 | "same": {Cursor: "same", Rows: []suite.FeedRow{row("a", 1, false)}}, |
| 696 | }, |
| 697 | bodies: map[string]string{"a": "Going nowhere."}, |
| 698 | } |
| 699 | sib, client := f.start(t) |
| 700 | s := storeForTest(t) |
| 701 | res, err := s.Poll(ctx, client, sib) |
| 702 | if err != nil { |
| 703 | t.Fatalf("poll: %v", err) |
| 704 | } |
| 705 | if !res.More { |
| 706 | t.Fatal("the fixture did not produce a short poll") |
| 707 | } |
| 708 | if at := polledAt(t, s, "Docs"); !at.IsZero() { |
| 709 | t.Errorf("a poll reporting More recorded PolledAt=%v", at) |
| 710 | } |
| 711 | }) |
| 712 | } |
| 713 | |
| 714 | // Two siblings in one store keep their own position and their own |
| 715 | // documents. |
| 716 | // |
| 717 | // Neither scoping had a fixture: a review dropped `source = ?` from the |
| 718 | // tombstone lookup and `app = ?` from the cursor read, and both stayed |
| 719 | // green. In production the first means Docs' tombstone for item "42" |
| 720 | // deletes Sheets' item "42"; the second means Sheets polls from Docs' |
| 721 | // cursor and silently skips everything before it. |
| 722 | // |
| 723 | // MUTATION-TESTED 2026-09-07, each verified red then reverted: |
| 724 | // |
| 725 | // deleteBySiblingItem without `source = ?` |
| 726 | // -> red, the other sibling's document goes |
| 727 | // cursor()/upsertFeed without `app = ?` |
| 728 | // -> red, the positions collide |
| 729 | func TestTwoSiblingsKeepTheirOwnPositionAndDocuments(t *testing.T) { |
| 730 | ctx := context.Background() |
| 731 | s := storeForTest(t) |
| 732 | |
| 733 | // The same item id in both apps, which is the collision that matters: |
| 734 | // item ids are the siblings' own and nothing makes them unique across |
| 735 | // apps. |
| 736 | docs := &feedServer{ |
| 737 | t: t, |
| 738 | pages: map[string]suite.Feed{ |
| 739 | "": {Cursor: "docs-1", Rows: []suite.FeedRow{row("42", 1, false)}}, |
| 740 | "docs-1": {Cursor: "docs-1", Rows: []suite.FeedRow{}}, |
| 741 | }, |
| 742 | bodies: map[string]string{"42": "Docs' document forty-two."}, |
| 743 | } |
| 744 | sheets := &feedServer{ |
| 745 | t: t, |
| 746 | pages: map[string]suite.Feed{ |
| 747 | "": {Cursor: "sheets-9", Rows: []suite.FeedRow{row("42", 1, false)}}, |
| 748 | "sheets-9": {Cursor: "sheets-9", Rows: []suite.FeedRow{}}, |
| 749 | }, |
| 750 | bodies: map[string]string{"42": "Sheets' document forty-two."}, |
| 751 | } |
| 752 | dSib, client := docs.start(t) |
| 753 | sSib, _ := sheets.start(t) |
| 754 | sSib.Name = "Sheets" |
| 755 | |
| 756 | if _, err := s.Poll(ctx, client, dSib); err != nil { |
| 757 | t.Fatalf("poll Docs: %v", err) |
| 758 | } |
| 759 | if _, err := s.Poll(ctx, client, sSib); err != nil { |
| 760 | t.Fatalf("poll Sheets: %v", err) |
| 761 | } |
| 762 | |
| 763 | // Separate positions, or one app resumes from the other's place. |
| 764 | if got := cursorOf(t, s, "Docs"); got != "docs-1" { |
| 765 | t.Errorf("Docs cursor = %q, want docs-1", got) |
| 766 | } |
| 767 | if got := cursorOf(t, s, "Sheets"); got != "sheets-9" { |
| 768 | t.Errorf("Sheets cursor = %q, want sheets-9", got) |
| 769 | } |
| 770 | assertHeld(t, s, "Docs", "42", 1) |
| 771 | assertHeld(t, s, "Sheets", "42", 1) |
| 772 | |
| 773 | // Docs tombstones its "42". Sheets' must survive. |
| 774 | docs.pages["docs-1"] = suite.Feed{Cursor: "docs-2", Rows: []suite.FeedRow{row("42", 2, true)}} |
| 775 | docs.pages["docs-2"] = suite.Feed{Cursor: "docs-2", Rows: []suite.FeedRow{}} |
| 776 | res, err := s.Poll(ctx, client, dSib) |
| 777 | if err != nil { |
| 778 | t.Fatalf("tombstone poll: %v", err) |
| 779 | } |
| 780 | if res.Deleted != 1 { |
| 781 | t.Errorf("result = %+v, want 1 deleted", res) |
| 782 | } |
| 783 | if heldCount(t, s, "Docs") != 0 { |
| 784 | t.Error("Docs' document survived its own tombstone") |
| 785 | } |
| 786 | assertHeld(t, s, "Sheets", "42", 1) |
| 787 | } |
| 788 | |
| 789 | // A tombstone memoria has already applied deletes nothing and says so. |
| 790 | // |
| 791 | // A feed can legitimately re-announce a deletion, and a poll that failed |
| 792 | // mid-page re-applies the rows before the failure by design — so the count |
| 793 | // has to be of documents that actually went, not of rows that mentioned |
| 794 | // one. Counting rows makes the sources page report deletions that did not |
| 795 | // happen, on exactly the days something went wrong. |
| 796 | // |
| 797 | // MUTATION-TESTED 2026-09-07: `res.Deleted++` per tombstone row rather |
| 798 | // than per document removed. Red. Reverted. |
| 799 | func TestAReAnnouncedTombstoneDeletesNothing(t *testing.T) { |
| 800 | ctx := context.Background() |
| 801 | f := &feedServer{ |
| 802 | t: t, |
| 803 | pages: map[string]suite.Feed{ |
| 804 | "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 2, true)}}, |
| 805 | "c1": {Cursor: "c1", Rows: []suite.FeedRow{}}, |
| 806 | }, |
| 807 | bodies: map[string]string{}, |
| 808 | } |
| 809 | sib, client := f.start(t) |
| 810 | s := storeForTest(t) |
| 811 | |
| 812 | // memoria never held "a" at all. |
| 813 | res, err := s.Poll(ctx, client, sib) |
| 814 | if err != nil { |
| 815 | t.Fatalf("poll: %v", err) |
| 816 | } |
| 817 | if res.Deleted != 0 { |
| 818 | t.Errorf("result = %+v: a tombstone for a document memoria does not "+ |
| 819 | "hold removed nothing, and the count must say so", res) |
| 820 | } |
| 821 | } |
| 822 | |
| 823 | // A sibling answering 403 on S3 is a contract breach, not a missing item, |
| 824 | // and it must not be quietly folded into the count that means "gone or |
| 825 | // not visible". |
| 826 | // |
| 827 | // 403 says "this exists and you may not have it" — the one fact custody |
| 828 | // hides — so a sibling emitting one is leaking, and the module keeps it |
| 829 | // distinguishable on purpose. Counting it as Missing would hide a leaking |
| 830 | // sibling behind a number that means something else. |
| 831 | // |
| 832 | // MUTATION-TESTED 2026-09-07: fold ErrContractBreach into the Missing |
| 833 | // branch. Red. Reverted. |
| 834 | func TestA403OnS3IsNotAMissingItem(t *testing.T) { |
| 835 | f := &feedServer{ |
| 836 | t: t, |
| 837 | pages: map[string]suite.Feed{ |
| 838 | "": {Cursor: "c1", Rows: []suite.FeedRow{row("a", 1, false)}}, |
| 839 | "c1": {Cursor: "c1", Rows: []suite.FeedRow{}}, |
| 840 | }, |
| 841 | bodies: map[string]string{}, |
| 842 | forbidS3: map[string]bool{"a": true}, |
| 843 | } |
| 844 | sib, client := f.start(t) |
| 845 | s := storeForTest(t) |
| 846 | |
| 847 | res, err := s.Poll(context.Background(), client, sib) |
| 848 | if err == nil { |
| 849 | t.Fatalf("a 403 from S3 came back as a completed poll: %+v", res) |
| 850 | } |
| 851 | if res.Missing != 0 { |
| 852 | t.Errorf("a 403 was counted as a missing item: %+v — that number "+ |
| 853 | "means 'gone or not team-visible', and a 403 means the sibling "+ |
| 854 | "is telling callers which ids exist", res) |
| 855 | } |
| 856 | } |
| 857 | |
| 858 | func documentOf(t *testing.T, s *Store, app, itemID string) Document { |
| 859 | t.Helper() |
| 860 | external, err := s.externalID(itemID) |
| 861 | if err != nil { |
| 862 | t.Fatalf("externalID: %v", err) |
| 863 | } |
| 864 | var d Document |
| 865 | if err := s.db.G.Where("source = ? AND external_id = ?", app, external).First(&d).Error; err != nil { |
| 866 | t.Fatalf("%s/%s not held: %v", app, itemID, err) |
| 867 | } |
| 868 | return d |
| 869 | } |
| 870 | |
| 871 | func polledAt(t *testing.T, s *Store, app string) time.Time { |
| 872 | t.Helper() |
| 873 | var row SiblingFeed |
| 874 | if err := s.db.G.Where("app = ?", app).First(&row).Error; err != nil { |
| 875 | return time.Time{} |
| 876 | } |
| 877 | return row.PolledAt |
| 878 | } |
| 879 | |
amadan