| 1 | package sibling |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "sync" |
| 6 | |
| 7 | "amadan.net/oficina/suite" |
| 8 | ) |
| 9 | |
| 10 | // A Candidate is one passage the ranker proposes, carrying the app of |
| 11 | // record that has to rule on it before anybody reads it. |
| 12 | // |
| 13 | // The one rule, at the only place it can be enforced: **memoria never |
| 14 | // speaks a passage without asking whoever owns it whether this person may |
| 15 | // see it, right now.** Every path that would render a passage goes through |
| 16 | // Rule below, so there is no passage-only route that skips the question. |
| 17 | type Candidate struct { |
| 18 | Sibling Sibling |
| 19 | Item suite.ItemRevision |
| 20 | } |
| 21 | |
| 22 | // Rule asks every contributing sibling about its own candidates, **all of |
| 23 | // them at once**, and answers for every candidate in the order given. |
| 24 | // |
| 25 | // §6.5, and the number that matters is not the deadline but the shape: |
| 26 | // *"One page of results is one round of calls, so the page's floor is the |
| 27 | // slowest sibling, not their sum."* Three siblings at 800 ms each is a |
| 28 | // page that waits 800 ms, and a page that waits 2.4 s is a bug that no |
| 29 | // per-call assertion can see — which is why gate 23 holds the deadline and |
| 30 | // the concurrency together. Either alone passes on the other's failure. |
| 31 | // |
| 32 | // **Identical pairs are asked once.** A document contributes many chunks |
| 33 | // and every one of them carries the same `(item_id, revision)`, so the |
| 34 | // obvious loop asks a ten-chunk document about ten times over. That is not |
| 35 | // merely wasteful: `DecodeS2` maps a duplicated row to UNKNOWN, so a |
| 36 | // sibling answering the same pair once for a question that named it ten |
| 37 | // times is a batch that partly withholds itself. Dedupe first, fan the |
| 38 | // answers back out. |
| 39 | // |
| 40 | // **A sibling with more candidates than one batch allows gets several |
| 41 | // concurrent batches**, for the same reason the siblings are concurrent: a |
| 42 | // page whose cost grows with the number of batches is a page whose cost |
| 43 | // grows with the corpus. Gate 19's mixed-provenance artefact is the case |
| 44 | // that spans them. |
| 45 | // |
| 46 | // Every candidate gets exactly one Disposition. There is no error return |
| 47 | // and no short slice: a caller with a fifth outcome to handle has somewhere |
| 48 | // to forget to handle it, and the forgotten branch renders. |
| 49 | // |
| 50 | // **It returns dispositions and not notices**, and it used to return both. |
| 51 | // The caller that renders a page has to aggregate in ITS order anyway — |
| 52 | // the reader meets the notices next to the results they are about — so the |
| 53 | // Outcome built here was recomputed and discarded, which is a value nobody |
| 54 | // uses sitting on a concurrent code path. A review then showed what that |
| 55 | // costs: moving the recording inside the goroutines is an ordinary |
| 56 | // refactor, it left every test green, and it is a concurrent map write — |
| 57 | // `fatal error` on §6.5's most ordinary case, a cold wake where two |
| 58 | // siblings both time out. Deleting the return removes the place that bug |
| 59 | // can live rather than adding a test that watches for it. |
| 60 | func (c *Client) Rule(ctx context.Context, asker suite.Subject, candidates []Candidate) []Disposition { |
| 61 | out := make([]Disposition, len(candidates)) |
| 62 | if len(candidates) == 0 { |
| 63 | return out |
| 64 | } |
| 65 | |
| 66 | // Group by sibling, deduping pairs as we go. positions maps a pair back |
| 67 | // to every candidate index that asked about it, which is what makes one |
| 68 | // answer serve ten chunks. |
| 69 | type work struct { |
| 70 | sib Sibling |
| 71 | pairs []suite.ItemRevision |
| 72 | positions map[suite.ItemRevision][]int |
| 73 | } |
| 74 | byS := map[Sibling]*work{} |
| 75 | var order []Sibling |
| 76 | for i, cand := range candidates { |
| 77 | w, ok := byS[cand.Sibling] |
| 78 | if !ok { |
| 79 | w = &work{sib: cand.Sibling, positions: map[suite.ItemRevision][]int{}} |
| 80 | byS[cand.Sibling] = w |
| 81 | order = append(order, cand.Sibling) |
| 82 | } |
| 83 | if _, asked := w.positions[cand.Item]; !asked { |
| 84 | w.pairs = append(w.pairs, cand.Item) |
| 85 | } |
| 86 | w.positions[cand.Item] = append(w.positions[cand.Item], i) |
| 87 | } |
| 88 | |
| 89 | // One goroutine per batch, every batch in flight together. The answers |
| 90 | // are written into `out` at indices owned by exactly one goroutine — |
| 91 | // a pair belongs to one sibling and each sibling's batches partition |
| 92 | // its pairs — so no lock guards `out`. |
| 93 | var wg sync.WaitGroup |
| 94 | for _, sib := range order { |
| 95 | w := byS[sib] |
| 96 | for _, batch := range batches(w.pairs) { |
| 97 | wg.Add(1) |
| 98 | go func(sib Sibling, batch []suite.ItemRevision, positions map[suite.ItemRevision][]int) { |
| 99 | defer wg.Done() |
| 100 | scatter(batch, c.Authorize(ctx, sib, asker, batch), positions, out) |
| 101 | }(w.sib, batch, w.positions) |
| 102 | } |
| 103 | } |
| 104 | wg.Wait() |
| 105 | return out |
| 106 | } |
| 107 | |
| 108 | // scatter writes one batch's answers out to every candidate that asked |
| 109 | // about each pair — the ten chunks of one document that produced a single |
| 110 | // question. |
| 111 | // |
| 112 | // **It is a function rather than four lines inside the goroutine because |
| 113 | // its short-slice arm is otherwise unreachable, and an unreachable guard is |
| 114 | // one written to pass.** `Authorize` returns exactly one answer per pair |
| 115 | // asked, always, so nothing a test can stand up makes `answers` short. The |
| 116 | // mutation that proved it: change the fallback from UNKNOWN to VISIBLE and |
| 117 | // every test stayed green, because the branch never ran. |
| 118 | // |
| 119 | // So the branch is reachable here, `TestAShortAnswerSliceWithholds` reaches |
| 120 | // it, and the invariant it is defending against a future change to — |
| 121 | // Authorize's totality — has its own test one file over. The alternative |
| 122 | // was to delete the arm and index blindly, which turns a contract change |
| 123 | // into a panic that takes the page down mid-render. |
| 124 | func scatter(batch []suite.ItemRevision, answers []suite.Authorization, positions map[suite.ItemRevision][]int, out []Disposition) { |
| 125 | for j, pair := range batch { |
| 126 | // A missing answer is an answer memoria does not have, which is |
| 127 | // UNKNOWN, which withholds loudly. Never Render, and never the |
| 128 | // silent omission — a page cannot go quiet because a slice was short. |
| 129 | a := suite.UNKNOWN |
| 130 | if j < len(answers) { |
| 131 | a = answers[j] |
| 132 | } |
| 133 | d := Decide(a) |
| 134 | for _, idx := range positions[pair] { |
| 135 | out[idx] = d |
| 136 | } |
| 137 | } |
| 138 | } |
| 139 | |
| 140 | // batches splits a sibling's pairs into calls the module will accept. |
| 141 | // |
| 142 | // The cap and its reason are the module's (`MaxS2Pairs`). Splitting rather |
| 143 | // than truncating is `Authorize`'s rule one level up: a truncated question |
| 144 | // answers about fewer items than were asked, and the pairs that fell off |
| 145 | // the end come back as no answer at all — which withholds, and looks like |
| 146 | // a sibling problem when it is ours. |
| 147 | func batches(pairs []suite.ItemRevision) [][]suite.ItemRevision { |
| 148 | var out [][]suite.ItemRevision |
| 149 | for len(pairs) > suite.MaxS2Pairs { |
| 150 | out = append(out, pairs[:suite.MaxS2Pairs]) |
| 151 | pairs = pairs[suite.MaxS2Pairs:] |
| 152 | } |
| 153 | if len(pairs) > 0 { |
| 154 | out = append(out, pairs) |
| 155 | } |
| 156 | return out |
| 157 | } |
| 158 | |
amadan