| 1 | package memoria |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "strings" |
| 8 | "time" |
| 9 | |
| 10 | "amadan.net/oficina/suite" |
| 11 | |
| 12 | "amadan.net/oficina/memoria/internal/memoria/seal" |
| 13 | "amadan.net/oficina/memoria/internal/memoria/sibling" |
| 14 | ) |
| 15 | |
| 16 | // maxFeedPages bounds one poll. |
| 17 | // |
| 18 | // A sibling whose cursor never advances, or whose feed loops, must not |
| 19 | // hold a wake open forever. A ceiling on damage rather than a |
| 20 | // measurement: at a few hundred rows a page this is far past any ordinary |
| 21 | // catch-up, and reaching it is reported as an INCOMPLETE poll rather than |
| 22 | // as a finished one — the difference between "there is no more" and "I |
| 23 | // stopped looking" is the difference this whole file is about. |
| 24 | const maxFeedPages = 50 |
| 25 | |
| 26 | // PollResult is what one poll did. Counts only: no item id, no title, no |
| 27 | // body, so it can go in a log line or on the sources page (AGENTS.md). |
| 28 | type PollResult struct { |
| 29 | Stored int |
| 30 | Deleted int |
| 31 | // Missing counts rows the sibling listed and then would not serve. |
| 32 | // |
| 33 | // **Not an error, and emphatically not a deletion.** §4.2 makes S3's |
| 34 | // 404 cover both "gone" and "not team-visible", deliberately, so |
| 35 | // memoria cannot tell which — and deleting on it would let a |
| 36 | // visibility change erase history that the sibling still has. What it |
| 37 | // feeds is §8.1's S3 breaker, which is the thing that notices when |
| 38 | // this happens wholesale. |
| 39 | Missing int |
| 40 | // Empty counts items the sibling served with no body at all. |
| 41 | // |
| 42 | // Skipped, not stored: a stub page has no passages, so there is |
| 43 | // nothing to retrieve and nothing to cite. Counted rather than passed |
| 44 | // over in silence, because a source that is suddenly all stubs is a |
| 45 | // fact about the sibling worth being able to see. |
| 46 | Empty int |
| 47 | // More means the poll stopped before the sibling ran out. What |
| 48 | // memoria holds is consistent; it is not current, and the sources |
| 49 | // page must not say it is. |
| 50 | More bool |
| 51 | } |
| 52 | |
| 53 | // Poll brings memoria's copy of one sibling's content up to date. |
| 54 | // |
| 55 | // **The cursor advances only after a page has been fully applied**, and |
| 56 | // that ordering is the whole design. §5.1 and gate 7: a timed-out, |
| 57 | // rate-limited or half-applied enumeration returns no information rather |
| 58 | // than no documents, and a cursor moved past a page nobody applied skips |
| 59 | // changes the sibling will never offer again — silently, because the |
| 60 | // sibling has no way to know. Every failure below returns before the |
| 61 | // cursor is stored. |
| 62 | // |
| 63 | // **A failed poll deletes nothing**, which is the same rule from the other |
| 64 | // end. Deletions arrive as `gone` rows on a page that was fetched |
| 65 | // successfully; they are never inferred from an item's absence, because to |
| 66 | // a consumer that infers, a short page and a complete one look the same. |
| 67 | // |
| 68 | // Re-applying is the cheap failure and is chosen deliberately. When a row |
| 69 | // fails mid-page the cursor stays put, so the whole page comes back next |
| 70 | // time and its earlier rows are applied again — which is safe, since |
| 71 | // Ingest is keyed on (source, external id) and deleting an absent row is a |
| 72 | // no-op. The expensive failure is skipping a row nobody will mention |
| 73 | // again. |
| 74 | func (s *Store) Poll(ctx context.Context, c *sibling.Client, sib sibling.Sibling) (PollResult, error) { |
| 75 | var res PollResult |
| 76 | cursor, err := s.cursor(ctx, sib.Name) |
| 77 | if err != nil { |
| 78 | return res, err |
| 79 | } |
| 80 | |
| 81 | for page := 0; ; page++ { |
| 82 | if page >= maxFeedPages { |
| 83 | // Neither an error nor a completed poll: what memoria has is |
| 84 | // consistent, and there is more. Returning nil with More unset |
| 85 | // would let the caller call the corpus current. |
| 86 | res.More = true |
| 87 | return res, nil |
| 88 | } |
| 89 | feed, err := c.Feed(ctx, sib, cursor) |
| 90 | if err != nil { |
| 91 | return res, fmt.Errorf("poll %s: %w", sib.Name, err) |
| 92 | } |
| 93 | |
| 94 | for _, row := range feed.Rows { |
| 95 | if err := s.applyRow(ctx, c, sib, row, &res); err != nil { |
| 96 | return res, fmt.Errorf("poll %s: %w", sib.Name, err) |
| 97 | } |
| 98 | } |
| 99 | |
| 100 | // The page is applied. Only now may the position move. |
| 101 | if feed.Cursor != "" && feed.Cursor != cursor { |
| 102 | if err := s.setCursor(ctx, sib.Name, feed.Cursor); err != nil { |
| 103 | return res, err |
| 104 | } |
| 105 | cursor = feed.Cursor |
| 106 | continue |
| 107 | } |
| 108 | |
| 109 | // The cursor did not move. If the page was empty, the sibling |
| 110 | // answered and has nothing further — **the one place an empty |
| 111 | // result may be read as "nothing", and only because Feed returns |
| 112 | // an error for every other reason it could be empty.** If the page |
| 113 | // had rows, a cursor that does not advance would loop forever, so |
| 114 | // stop and say the poll did not conclude. |
| 115 | res.More = len(feed.Rows) > 0 |
| 116 | if !res.More { |
| 117 | if err := s.markPolled(ctx, sib.Name); err != nil { |
| 118 | return res, err |
| 119 | } |
| 120 | } |
| 121 | return res, nil |
| 122 | } |
| 123 | } |
| 124 | |
| 125 | func (s *Store) applyRow(ctx context.Context, c *sibling.Client, sib sibling.Sibling, row suite.FeedRow, res *PollResult) error { |
| 126 | // **There is no check for an empty item id here, and that is |
| 127 | // deliberate.** One was written; it could not fail. Every row reaching |
| 128 | // this function came out of `suite.DecodeS1`, which refuses the whole |
| 129 | // page — "S1 row %d has no item_id" (s1.go) — so the guard never ran, |
| 130 | // and a mutation deleting it left the suite green because the module |
| 131 | // was doing the work. A defence nothing can reach is a guard written |
| 132 | // to pass, whatever the intent. |
| 133 | // |
| 134 | // What memoria depends on is the module's refusal, and what makes that |
| 135 | // dependency visible is a test asserting the MODULE behaves that way |
| 136 | // rather than a duplicate check here pretending to. |
| 137 | if row.Gone { |
| 138 | n, err := s.deleteBySiblingItem(ctx, sib.Name, row.ItemID) |
| 139 | if err != nil { |
| 140 | return err |
| 141 | } |
| 142 | res.Deleted += n |
| 143 | return nil |
| 144 | } |
| 145 | |
| 146 | body, err := c.Content(ctx, sib, row.ItemID, row.Revision) |
| 147 | switch { |
| 148 | case errors.Is(err, suite.ErrNotAvailable): |
| 149 | // Listed, then not served. See PollResult.Missing: counted, not |
| 150 | // deleted, not failed. |
| 151 | res.Missing++ |
| 152 | return nil |
| 153 | case err != nil: |
| 154 | return err |
| 155 | } |
| 156 | |
| 157 | // **An empty item is skipped and counted, not stored and not failed.** |
| 158 | // |
| 159 | // This said `AllowTitleOnly: true` and claimed it stopped one empty |
| 160 | // document blocking a catch-up. It did the opposite: no Title is |
| 161 | // passed here — a sibling's S3 gives a body and nothing else — so |
| 162 | // Ingest refused with "neither a body nor a title", the row failed, |
| 163 | // the cursor stayed put, and **every subsequent poll of that sibling |
| 164 | // failed on the same row forever**. One stub page in Docs and nothing |
| 165 | // from Docs is ever ingested again. A review found it. |
| 166 | // |
| 167 | // Skipping is right rather than a fallback title: a stub page has no |
| 168 | // passages, so there is nothing to retrieve and nothing to cite, and |
| 169 | // inventing a title from an empty body would put a document in the |
| 170 | // corpus whose only content is its own name. The count is what keeps |
| 171 | // it from being silent. |
| 172 | if strings.TrimSpace(string(body)) == "" { |
| 173 | res.Empty++ |
| 174 | return nil |
| 175 | } |
| 176 | |
| 177 | // The item id is the stable key, so ExternalID's fingerprint is |
| 178 | // computed over it. A title would not be: a rename would make a second |
| 179 | // document and leave the first indexed under the old name. |
| 180 | _, err = s.Ingest(ctx, IngestSpec{ |
| 181 | Source: sib.Name, |
| 182 | Key: row.ItemID, |
| 183 | Body: string(body), |
| 184 | ItemID: row.ItemID, |
| 185 | SiblingRevision: row.Revision, |
| 186 | }) |
| 187 | switch { |
| 188 | case errors.Is(err, ErrIndexLag): |
| 189 | // Stored, and only the index is behind. The document is in the |
| 190 | // database, so the cursor may pass this row: re-fetching it would |
| 191 | // not fix the index either, and the app says the lag out loud. |
| 192 | case err != nil: |
| 193 | return err |
| 194 | } |
| 195 | res.Stored++ |
| 196 | return nil |
| 197 | } |
| 198 | |
| 199 | // cursor reads memoria's position in one sibling's feed. |
| 200 | // |
| 201 | // **A cursor that will not open is an error, never an empty cursor.** An |
| 202 | // empty cursor means "start from the beginning", which is a legitimate and |
| 203 | // very different instruction: it would re-announce the sibling's entire |
| 204 | // corpus, and — worse — it is exactly what a corrupted row would silently |
| 205 | // turn into. |
| 206 | func (s *Store) cursor(ctx context.Context, app string) (string, error) { |
| 207 | var row SiblingFeed |
| 208 | err := s.db.G.WithContext(ctx).Where("app = ?", app).First(&row).Error |
| 209 | switch { |
| 210 | case errors.Is(err, errRecordNotFound): |
| 211 | // No row yet: the first poll of a newly configured sibling starts |
| 212 | // from the beginning, which is what an empty cursor asks for. |
| 213 | return "", nil |
| 214 | case err != nil: |
| 215 | return "", err |
| 216 | } |
| 217 | if len(row.SealedCursor) == 0 { |
| 218 | return "", nil |
| 219 | } |
| 220 | raw, err := s.sealer.Open(seal.PurposeCursor, row.SealedCursor) |
| 221 | if err != nil { |
| 222 | return "", fmt.Errorf("open cursor for %s: %w", app, err) |
| 223 | } |
| 224 | return string(raw), nil |
| 225 | } |
| 226 | |
| 227 | // setCursor stores a position. **It does not touch PolledAt.** |
| 228 | // |
| 229 | // The two used to move together, and that was wrong in the direction that |
| 230 | // matters. A multi-page catch-up calls this once per page, so a poll that |
| 231 | // failed on page four — or one that stopped at the page bound with more to |
| 232 | // come — left PolledAt at the moment of the last page it managed. The |
| 233 | // sources page would then report the corpus current as of a time when it |
| 234 | // demonstrably was not, which is §8.1's rule inverted: *"a failed or stale |
| 235 | // S1 → the page states the corpus is current as of a time, rather than |
| 236 | // implying freshness."* |
| 237 | // |
| 238 | // A position is "how far I got". A timestamp is "and there was nothing |
| 239 | // after it". Only the second is a claim about completeness, and only the |
| 240 | // drained-feed branch of Poll may make it. |
| 241 | func (s *Store) setCursor(ctx context.Context, app, cursor string) error { |
| 242 | sealed, err := s.sealer.Seal(seal.PurposeCursor, []byte(cursor)) |
| 243 | if err != nil { |
| 244 | return err |
| 245 | } |
| 246 | return s.upsertFeed(ctx, app, map[string]any{"sealed_cursor": sealed}) |
| 247 | } |
| 248 | |
| 249 | // markPolled records that a poll reached the end of the feed without |
| 250 | // moving the cursor — the sibling answered and had nothing new. |
| 251 | func (s *Store) markPolled(ctx context.Context, app string) error { |
| 252 | return s.upsertFeed(ctx, app, map[string]any{"polled_at": time.Now().UTC()}) |
| 253 | } |
| 254 | |
| 255 | func (s *Store) upsertFeed(ctx context.Context, app string, fields map[string]any) error { |
| 256 | return s.db.G.WithContext(ctx).Transaction(func(tx *gormTx) error { |
| 257 | var row SiblingFeed |
| 258 | err := tx.Where("app = ?", app).First(&row).Error |
| 259 | switch { |
| 260 | case errors.Is(err, errRecordNotFound): |
| 261 | row = SiblingFeed{App: app} |
| 262 | if v, ok := fields["sealed_cursor"].([]byte); ok { |
| 263 | row.SealedCursor = v |
| 264 | } |
| 265 | if v, ok := fields["polled_at"].(time.Time); ok { |
| 266 | row.PolledAt = v |
| 267 | } |
| 268 | return tx.Create(&row).Error |
| 269 | case err != nil: |
| 270 | return err |
| 271 | } |
| 272 | // Scoped to tx, never the outer handle (AGENTS.md). |
| 273 | return tx.Model(&SiblingFeed{}).Where("id = ?", row.ID).Updates(fields).Error |
| 274 | }) |
| 275 | } |
| 276 | |
| 277 | // deleteBySiblingItem applies one `gone` row. |
| 278 | // |
| 279 | // It resolves the sibling's item id the same way Ingest stores it — the |
| 280 | // keyed fingerprint over (source, key) — because that is the only handle |
| 281 | // memoria has on a row from the outside. Returns how many documents went, |
| 282 | // which is 0 for a tombstone memoria has already applied or never held: |
| 283 | // **not an error.** A feed can legitimately re-announce a deletion, and a |
| 284 | // poll that failed mid-page re-applies the rows before the failure by |
| 285 | // design. |
| 286 | func (s *Store) deleteBySiblingItem(ctx context.Context, app, itemID string) (int, error) { |
| 287 | external, err := s.externalID(itemID) |
| 288 | if err != nil { |
| 289 | return 0, err |
| 290 | } |
| 291 | var docs []Document |
| 292 | if err := s.db.G.WithContext(ctx). |
| 293 | Where("source = ? AND external_id = ?", app, external). |
| 294 | Find(&docs).Error; err != nil { |
| 295 | return 0, err |
| 296 | } |
| 297 | for _, d := range docs { |
| 298 | // Store.Delete removes the index entries BEFORE the rows and |
| 299 | // refuses the whole delete if that fails, so a tombstone cannot |
| 300 | // leave passages indexed under a document that is gone. That |
| 301 | // ordering is AGENTS.md's rule and the reason this does not do |
| 302 | // its own deleting. |
| 303 | if err := s.Delete(ctx, d.ID); err != nil { |
| 304 | return 0, err |
| 305 | } |
| 306 | } |
| 307 | return len(docs), nil |
| 308 | } |
| 309 | |
| 310 | // polledAt reports when this sibling's feed was last drained. |
| 311 | // |
| 312 | // **A read that fails is an error, not a zero time.** A zero time means |
| 313 | // "never checked", which the page renders as its own sentence; returning |
| 314 | // it for a failed read would put a false one on the page — the same rule |
| 315 | // as everywhere else, on the value the freshness claim is built from. |
| 316 | func (s *Store) polledAt(ctx context.Context, app string) (time.Time, error) { |
| 317 | var row SiblingFeed |
| 318 | err := s.db.G.WithContext(ctx).Where("app = ?", app).First(&row).Error |
| 319 | switch { |
| 320 | case errors.Is(err, errRecordNotFound): |
| 321 | return time.Time{}, nil |
| 322 | case err != nil: |
| 323 | return time.Time{}, err |
| 324 | } |
| 325 | return row.PolledAt, nil |
| 326 | } |
| 327 | |
amadan