Compare commits

..
Author SHA1 Message Date
jschoubben 40da368ee3 Date an adopted merge once, by the walk it was adopted from, so a later walk it follows never makes it look newer than a forge merge heard after it (issue 1726, review)
mesh/check pass 40da368ee34e on 25176e1ccdb3 | unit 31 run, 4404 kept | integration 0 run, 4477 kept | over bound: unit | composition passed in 1ms on…
mesh/delivery ready: it delivers once merged
2026-10-24 13:55:10 +02:00
jschoubben 15101c9ce6 Move source.main for a merge the gate caught up after a gap, so the checker re-checks open heads on the trunk that is (issue 1726)
The checker takes a repository's trunk head only from the gate's source.main. On 2026-10-24 gitea's
restarted poll announced mesh-catalog #471's merge while the gate was starting and could not yet tell
whether it held the seat; the gate acknowledged the event unread. It then adopted the controller's walk
of the merge commit, which carries no merge time, so every older merge of the repository looked newer
and source.main was never moved: #472 was checked on the trunk before #471 and its merge refused.

- An event heard before the gate can tell whether it holds the seat is offered again, not taken.
- An adopted merge is dated by its walk's creation, so it settles into source.main as a forge merge does.
- Each time the gate comes to hold the seat it weighs the newest merge of each repository against
  source.main once more, so a merge marked there and never written (as #471) is written; never backwards.
2026-10-24 13:41:56 +02:00
jschoubben 545c72bb59 Merge pull request 'gitea's watcher answers its handshake without waiting for the forge, so a slow forge at a tool-runner restart no longer kills its tools and events (issue 1726)' (#471) from fix/1726-gitea-handshake-not-on-forge into main
Reviewed-on: #471
2026-10-24 11:10:39 +00:00
jschoubben 7afcd0282d Set up the forge's webhook without holding the bundle's import, so a slow forge never costs gitea its handshake and its tools (issue 1726)
At 12:37 on 2026-10-24 node-tools on novox was replaced; gitea's watcher bundle awaited the webhook's set-up on the forge's admin API at the top of its import, outlasted the 10 s handshake and was killed, never to start again. gitea_commit_status, gitea_merge_pull_request and gitea_events were then served nowhere: the forge bridge's calls found no responders and no new pull request was announced.
2026-10-24 13:00:57 +02:00
jschoubben 25176e1ccd Merge pull request 'mesh-build-go: lint the unit level's directories as many at once as its tests, so a large repository's check ends within its bound (issue 1930)' (#470) from fix/1930-parallel-lint into main
Reviewed-on: #470
2026-10-24 00:16:55 +00:00
jschoubben 7e1f3de77a mesh-build-go: lint directories side by side like the tests, so a large repository's check stays inside its bound (issue 1930)
mesh/delivery delivered
mesh/check failed: mesh-build-go: the check of novox/mesh-catalog#470: the ask mvlnph6fd84a7b's step on shanks did not end within 30m0s of its start |…
2026-10-24 02:01:36 +02:00
jschoubben 4b1bed4b9d Merge pull request 'mesh-mail-reader: read messenger's sent_on as a list, so an unchanged set of mails is signalled once, not every 160 s (issue 1926)' (#469) from fix/1926-mail-signal-once into main
Reviewed-on: #469
2026-10-23 22:36:10 +00:00
jschoubben 25d69406b7 Merge pull request 'mesh-build-go: a failed test keeps its last lines and its cause, not its first fifteen (issue 1901, with mesh-delivery)' (#466) from fix/1901-rebased into main
Reviewed-on: #466
2026-10-23 22:35:57 +00:00
jschoubben f836c021e1 Merge pull request 'gitea: the checker's owed results are taken at once and written in the background, so the runner keeps the watch and every mesh/check reaches the forge (issue 1726)' (#468) from fix/1726-owed-watch-not-blocked into main
Reviewed-on: #468
2026-10-23 22:35:33 +00:00
jschoubben 450f0bbe79 Merge pull request 'The build modules pin mesh-sdk's list-form level digest, so a check of a repository with a package named token ends (issue 1930, with mesh-sdk and mesh-delivery)' (#467) from fix/1930-digest-shape into main
Reviewed-on: #467
2026-10-23 22:08:24 +00:00
jschoubben 5a7076a6a2 Take the checker's owed without waiting on the forge, so a large owed never times its watch out (issue 1726)
mesh/delivery delivered
mesh/check checking 5a7076a6a25e on f836c021e14f
The runtime gives a watch's current values 10 s and waits for each to be taken; a forge write per key made ~200 keys outlast it at every attempt, so after node-tools restarted on novox at 23:25:58 the watch never held and no verdict reached mesh/check. Writes now run in the background, a few at a time, each key's in revision order.
2026-10-23 23:52:52 +02:00
jschoubben 30b22c198e Pin the build modules to mesh-sdk 403a5d7, whose level digest is a list, so a check of a repository with a package named token ends (issue 1930)
mesh/delivery failed: its walk failed, carrying its own commit: a gate on a first machine (what it carried put back), a build, a machine
mesh/delivery-group group fix/1930-digest-shape failed: a member failed
mesh/check checking 30b22c198e6a on 450f0bbe7984 | novox/mesh-catalog#467 in group fix/1930-digest-shape
2026-10-23 23:36:31 +02:00
jschoubben 753e8fc202 mesh-mail-reader: read tell's sent_on as the list it is, so a told set is not sent again; leave past-due and duplicate mails out (issue 1926)
mesh/delivery delivered
mesh/check checking 753e8fc202c3 on 4b1bed4b9d5b
2026-10-23 23:27:23 +02:00
mesh-tools e5202445ba Merge pull request 'mesh-build-go: build_go_test takes packages as a typed list too, so checks run again (issue 1925)' (#465) from fix/1925-packages-shape into main 2026-10-23 20:44:09 +00:00
jschoubben dcae699323 mesh-build-go: a clipped failure starts with its cause, not the marker, for an older checker (issue 1901/AC3)
mesh/delivery covered: a later merge that contains it was delivered: novox/mesh-catalog@753e8fc202c3 (merged as 4b1bed4b into main, walk plan-17927949853…
mesh/delivery-group group fix/1901-rebased delivered: every member is delivered
mesh/check checking dcae699323c8 on 25d69406b740 | novox/mesh-catalog#466 in group fix/1901-rebased
2026-10-23 22:26:29 +02:00
jschoubben 70ef99595a mesh-build-go: build_go_test takes packages as a typed list too (issue 1925)
mesh/delivery delivered
mesh/check failed: shanks: the check of novox/mesh-catalog#465 failed: the unit level: packages is a list of package patterns | 70ef99595aa9 on e52024…
2026-10-23 22:21:41 +02:00
jschoubben 746eb885a3 Test 1901/AC3: an older pair of build module and checker still agree (issue 1901) 2026-10-23 22:14:28 +02:00
jschoubben 2cdb9b8fc4 mesh-build-go: a failed test keeps its last lines and its error line, not its first fifteen (issue 1901) 2026-10-23 22:13:41 +02:00
jschoubben e830045545 Merge pull request 'R4g+R5 (mesh-catalog): the build modules pull asks, scripts become tests, build-agent references go (design 47)' (#464) from feat/r4g into main
Reviewed-on: #464
2026-10-23 17:49:43 +00:00
jschoubben 454fb8a918 Write a pull request's check verdict into its key after the key is created: the status event can be taken first (R4g, issue 1887)
mesh/delivery-group group feat/r4g failed: a member failed
mesh/delivery covered after its failure: a later walk that contains it was delivered: plan-1792777718506798022 (novox/mesh-catalog at e8300455 into main,…
mesh/check checking 454fb8a918ad on 3bec357968da | novox/mesh-catalog#464 in group feat/r4g
2026-10-23 19:42:54 +02:00
jschoubben c32474ee40 Judge the delivery-before-check order from the step records, not a poll: at capacity 1 the check starts the instant the delivery ends (R4g, issue 1887)
mesh/delivery superseded: a newer head of the same pull request
mesh/check failed: integration: mesh-delivery-gate/cmd/mesh-d… | c32474ee40f4 on 3bec357968da | judged by the candidate's mesh-build-go at 2e667ac15ce7
2026-10-23 19:19:45 +02:00
jschoubben f95f3f430e Give the delivery-ask integration test a bus of its own: the level shares one bus across directories, so a named stream there could be taken by another test (R4g, issue 1887) 2026-10-23 19:15:52 +02:00
31 changed files with 974 additions and 92 deletions
+28 -9
View File
@@ -83,6 +83,9 @@ export function checkStatusOf(change: OwedChange): { owner: string; repo: string
return { owner, repo, sha, revision, status };
}
/** How many mesh/check writes this module has at the forge at once. */
export const WRITES_AT_ONCE = 4;
/** The one writer of mesh/check: each owed revision of the checker's, in revision order per key. */
export class CheckStatuses {
/** The newest revision written of each key, and the newest seen; the oldest keys go past what any forge holds open. */
@@ -91,6 +94,9 @@ export class CheckStatuses {
/** The turn each key's writes wait on, and the asks again still running. */
private turns = new Map<string, Promise<unknown>>();
private again = new Set<Promise<void>>();
/** Writes at the forge at once, at most WRITES_AT_ONCE: a watch's start hands over every key together (issue 1726). */
private writing = 0;
private queued: (() => void)[] = [];
private forge: CheckForge;
private say: (line: string) => void;
private sleep: (ms: number) => Promise<void>;
@@ -123,7 +129,7 @@ export class CheckStatuses {
this.say(`[gitea] ${asked.owner}/${asked.repo}@${asked.sha.slice(0, 12)}: mesh/check at revision ${asked.revision} is older than ${newest}, already written, and is dropped`);
return "dropped";
}
await this.forge.setCommitStatus(asked.owner, asked.repo, asked.sha, asked.status);
await this.atTheForge(() => this.forge.setCommitStatus(asked.owner, asked.repo, asked.sha, asked.status));
this.remember(this.written, key, asked.revision);
return "written";
});
@@ -133,19 +139,32 @@ export class CheckStatuses {
return turn;
}
/** What the watch hands over: written, or when the forge refuses, asked again on a bound in the background until it is
* written or a newer revision of the key was seen. Never throws: a watch's handler that throws loses the change. */
/** What the watch hands over: taken at once and written in the background, or when the forge refuses, asked again on a
* bound until it is written or a newer revision of the key was seen. Never throws: a watch's handler that throws loses
* the change. It never waits on the forge (novox/hq issue 1726): the runtime hands a watch's current values one at a
* time, waits for each to be taken and gives the whole set 10 s; a forge write per key made a large owed outlast that
* bound at every attempt, so the watch never held and no verdict reached mesh/check. Each key's writes still take
* turns in revision order (apply). */
async take(change: OwedChange): Promise<void> {
const run = this.apply(change).then(() => {}, (err) => this.askAgain(change, err));
this.again.add(run);
void run.finally(() => this.again.delete(run));
}
/** Runs one forge write when fewer than WRITES_AT_ONCE are running, else after one ends. */
private async atTheForge<T>(write: () => Promise<T>): Promise<T> {
if (this.writing >= WRITES_AT_ONCE) await new Promise<void>((r) => this.queued.push(r));
else this.writing++;
try {
await this.apply(change);
} catch (err) {
const run = this.askAgain(change, err);
this.again.add(run);
void run.finally(() => this.again.delete(run));
return await write();
} finally {
const next = this.queued.shift();
if (next) next();
else this.writing--;
}
}
/** Resolves when no ask again is running. */
/** Resolves when no write or ask again is running. */
async settled(): Promise<void> {
while (this.again.size) await Promise.all([...this.again]);
}
+5 -1
View File
@@ -589,7 +589,11 @@ if (gitea) {
};
// **The forge's events first, the polls as the safety net** (novox/hq issue 729): while verified deliveries arrive the
// polls slow to every five minutes and announce only what no event said (Pace); otherwise they keep 30 s.
await listenForForge(client);
// **Not awaited** (novox/hq issue 1726): the receiver's set-up asks the forge's admin API, and the runtime kills a bundle
// whose import has not ended in its 10 s handshake and never starts it again; a forge slow to answer while every bundle
// of the node starts left gitea_commit_status, gitea_merge_pull_request and gitea_events served nowhere. Until it is
// listening, Pace keeps the polls at 30 s, so nothing is missed meanwhile; listenForForge says its own failure.
void listenForForge(client);
// Main takes no push and only the tools' account merges, in every repository, kept and not only when a token is
// minted: at the start and every ten minutes after, so a repository made later gets it too (issue 948, ADR 0398 §2).
// Only while the setting keep-main-rules is true (mainrules.ts); false, the default, leaves every repository's rule as set.
+50
View File
@@ -109,3 +109,53 @@ test("a bundle whose read of the checker's owed is granted late writes the lates
assert.match(got.statuses[0], /"state":"success".*pass at 4/);
assert.match(err, /mesh-delivery-checker\.owed could not be bound/, "the refusal is said");
});
// **The bundle's import does not wait on the forge** (novox/hq issue 1726, 2026-10-24 12:37). The webhook receiver's set-up
// asks the forge's admin API for its system webhook; awaited at the top of index.ts, a forge slow to answer (busy while
// every bundle of the node starts) held the import past node-tools' 10 s handshake, the bundle was killed and never
// started again, and gitea_commit_status, gitea_merge_pull_request and gitea_events were served nowhere: the forge
// bridge's calls found no responders and no new pull request was announced. Here the forge stand-in takes every
// request and never answers: the import resolves at once all the same, with the bundle's tools registered.
test("importing the bundle resolves, its tools registered, while the forge never answers the webhook's set-up", async () => {
const index = pathToFileURL(join(here, "..", "dist", "index.js")).href;
const script = `
const http = await import("node:http");
const fs = await import("node:fs");
const os = await import("node:os");
const path = await import("node:path");
const forge = http.createServer(() => {});
await new Promise((r) => forge.listen(0, "127.0.0.1", r));
const dir = fs.mkdtempSync(path.join(os.tmpdir(), "gitea-1726-"));
fs.writeFileSync(path.join(dir, "admin.password"), "pw");
process.env.MESH_GITEA_URL = "http://127.0.0.1:" + forge.address().port;
process.env.MESH_GITEA_STATE_DIR = dir;
process.env.MESH_GITEA_ADMIN_PASSWORD_FILE = path.join(dir, "admin.password");
process.env.MESH_GITEA_HOOK_PORT = "0";
process.env.MESH_GITEA_HOOK_HOST = "127.0.0.1";
const { useBroker } = await import("@novox/mesh-sdk/messaging");
useBroker(() => ({
subscribe: async () => () => {}, publish: async () => {}, request: async () => { throw new Error("no responders"); },
handle: async () => () => {}, close: async () => {},
}));
await import(${JSON.stringify(index)});
const { collectTools } = await import("@novox/mesh-sdk/tools");
console.log(JSON.stringify({ imported: true, tools: collectTools().flatMap((r) => r.tools.map((t) => t.name)) }));
process.exit(0);`;
const env = { ...process.env, MESH_GITEA_TOKEN: "x", MESH_GITEA_ADMIN_USER: "admin", MESH_GITEA_HOOK_URL: "http://127.0.0.1:1" };
for (const k of ["MESH_SERVED_MODULE", "MESH_GITEA_ADMIN_PASSWORD", "MESH_GITEA_CONFIG_FILE"]) delete env[k];
const child = spawn(process.execPath, ["--input-type=module", "-e", script], { cwd: join(here, ".."), env, stdio: ["ignore", "pipe", "pipe"] });
let out = "";
let err = "";
child.stdout.on("data", (d) => (out += d));
child.stderr.on("data", (d) => (err += d));
const ended = new Promise<void>((r) => child.on("exit", () => r()));
const timer = setTimeout(() => child.kill("SIGKILL"), 3000);
await ended;
clearTimeout(timer);
const line = out.split("\n").find((l) => l.startsWith("{"));
assert.ok(line, `the import did not resolve within 3 s while the forge did not answer; stdout: ${out.slice(0, 400)} stderr: ${err.slice(0, 600)}`);
const got = JSON.parse(line!) as { tools: string[] };
for (const t of ["gitea_events", "gitea_commit_status", "gitea_merge_pull_request"]) {
assert.ok(got.tools.includes(t), `${t} is registered: ${got.tools}`);
}
});
+42
View File
@@ -133,3 +133,45 @@ test("no tool writes mesh/check: gitea_check_status is gone and gitea_commit_sta
const index = readFileSync(new URL("../index.ts", import.meta.url), "utf8");
assert.ok(!/gitea_check_status|checkStatusTool/.test(index), "index.ts serves no gitea_check_status");
});
test("the watch's current values are taken without waiting on the forge, so a slow forge never times the watch out (issue 1726)", async () => {
// The tool runtime hands a watch's current values one at a time and waits for each to be taken; it gives the whole set
// 10 s (mesh-tools StateTimeout) and otherwise ends the watch. On 2026-10-23 the owed held ~200 keys, each written to the
// forge before the next was handed over: the set took longer than 10 s, the watch ended at every attempt, and no
// verdict reached the forge's mesh/check again.
const { CheckStatuses, watchOwed } = await import("../check.ts");
const sent: string[] = [];
let release: () => void = () => {};
const slow = new Promise<void>((r) => { release = r; });
const f = { setCommitStatus: async (_o: string, _r: string, s: string, st: any) => { await slow; sent.push(`${s.slice(0, 4)} ${st.state}`); } };
const checks = new CheckStatuses(f, () => {}, noSleep);
const keys = Array.from({ length: 200 }, (_, i) => key(i.toString(16).padStart(4, "0") + sha.slice(4), "novox/mesh-sdk", i + 1));
let handedOver = 0;
let attempts = 0;
const watch = async (handler: (c: any) => Promise<void>) => {
if (++attempts > 1) return; // a second attempt is not what this test waits for: the first must hold
// The runtime's bound on the current values: the forge has answered none of them yet.
const timedOut = new Promise<"timeout">((r) => setTimeout(() => r("timeout"), 200));
const delivered = (async () => { for (const [i, k] of keys.entries()) { await handler(put(k, owed(), i + 1, true)); handedOver++; } return "delivered" as const; })();
if (await Promise.race([delivered, timedOut]) === "timeout") throw new Error("the current values of mesh-delivery-checker.owed did not arrive in 10s");
};
const said: string[] = [];
await watchOwed(watch, checks, noSleep, (l) => said.push(l));
assert.equal(handedOver, 200, "every current value was taken before the forge answered");
assert.ok(!said.some((l) => /could not be bound/.test(l)), said.join(" | "));
assert.equal(sent.length, 0, "nothing is written before the forge answers");
release();
await checks.settled();
assert.equal(sent.length, 200, "every owed status is written once the forge answers");
});
test("the writes a watch's start hands over go to the forge a few at a time, never all at once (issue 1726)", async () => {
const { CheckStatuses, WRITES_AT_ONCE } = await import("../check.ts");
let running = 0, most = 0, done = 0;
const f = { setCommitStatus: async () => { running++; most = Math.max(most, running); await new Promise((r) => setTimeout(r, 1)); running--; done++; } };
const checks = new CheckStatuses(f, () => {}, noSleep);
for (let i = 0; i < 50; i++) await checks.take(put(key(i.toString(16).padStart(4, "0") + sha.slice(4), "novox/mesh-sdk", i + 1), owed(), i + 1, true));
await checks.settled();
assert.equal(done, 50);
assert.equal(most, WRITES_AT_ONCE);
});
@@ -4,7 +4,6 @@ package main
import (
"context"
"os"
"sync"
"testing"
"time"
@@ -14,16 +13,25 @@ import (
"git.novox.be/novox/mesh-sdk/go/buildstep"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/testbus"
)
// checkCommit is the commit the check ask fetches, apart from the delivery ask's, so the fetch order names which ran first.
const checkCommit = "89abcdef0123456789abcdef0123456789abcdef"
// 1887/AC-order, 1887/AC-pull against the real bus: a check ask published before a delivery ask runs after it, and at
// capacity 1 the second waits in the stream until the first ends.
//
// **A bus of its own.** The integration level runs every directory's tests at once on one bus, and this test names the
// build seat's stream and its fixed workers (SEAT_NODE_BUILD_GO and its workers) and deletes the stream at its end, so
// anything else on that bus, a puller or a leftover consumer, could hold or take its delivery ask and let the check
// ask run first. It starts from testbus (mesh-sdk, novox/hq ADR 0495 clause 5): a bus for this test alone, gone after it.
//
// **The order is read from the record.** At capacity 1 the puller starts the check the instant the delivery ask's step
// ends, well before any poll, so "the check key holds a state" is true on correct behaviour (issue 1887). The test
// waits for both asks to end and compares the step records' own Started and Ended times, and the fetch order by commit.
func TestTheModuleTakesADeliveryAskBeforeAnEarlierCheck(t *testing.T) {
url := os.Getenv("MESH_TEST_NATS")
if url == "" {
t.Skip("MESH_TEST_NATS names no bus: the integration level starts one")
}
nc, err := nats.Connect(url)
nc, err := nats.Connect(testbus.URL(t))
if err != nil {
t.Fatal(err)
}
@@ -40,6 +48,7 @@ func TestTheModuleTakesADeliveryAskBeforeAnEarlierCheck(t *testing.T) {
t.Cleanup(func() { _ = js.DeleteStream(context.Background(), "SEAT_NODE_BUILD_GO") })
check := deliveryAsk()
check.Kind = delivery.AskCheck
check.Build.Fetch["commit"] = checkCommit
first := deliveryAsk()
for _, a := range []delivery.StepAsk{check, first} {
raw, _ := delivery.EncodeStepAsk(a)
@@ -68,30 +77,36 @@ func TestTheModuleTakesADeliveryAskBeforeAnEarlierCheck(t *testing.T) {
p := m.Puller(steps, &buildstep.JetStreamAsks{JS: js, Language: delivery.AskGo})
p.PullWait = 200 * time.Millisecond
go p.Pull(ctx)
for _, a := range []delivery.StepAsk{first, check} {
key, _ := delivery.AskStepKey("ace", a.ID)
for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(10 * time.Millisecond) {
states.mu.Lock()
v := states.steps[key]
states.mu.Unlock()
if len(v) > 0 && v[len(v)-1].State != delivery.StepRunning {
break
}
if time.Now().After(deadline) {
t.Fatalf("the ask %s (%s) never ended", a.ID, a.Kind)
}
}
var ran map[string][]delivery.Step
for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(10 * time.Millisecond) {
ran = map[string][]delivery.Step{}
ended := 0
states.mu.Lock()
checkKey, _ := delivery.AskStepKey("ace", check.ID)
if a.ID == first.ID && len(states.steps[checkKey]) > 0 {
states.mu.Unlock()
t.Fatal("the check ask started before the delivery ask ended at capacity 1")
for _, a := range []delivery.StepAsk{first, check} {
key, _ := delivery.AskStepKey("ace", a.ID)
v := append([]delivery.Step(nil), states.steps[key]...)
ran[a.ID] = v
if len(v) > 0 && v[len(v)-1].State != delivery.StepRunning {
ended++
}
}
states.mu.Unlock()
if ended == 2 {
break
}
if time.Now().After(deadline) {
t.Fatalf("the asks never both ended: %v", ran)
}
}
// The check's step may start the moment the delivery's ends, so the order is read from the records' own times.
deliveryEnded := ran[first.ID][len(ran[first.ID])-1].Ended
checkStarted := ran[check.ID][0].Started
if checkStarted.Before(deliveryEnded) {
t.Fatalf("the check ask started at %v, before the delivery ask ended at %v at capacity 1", checkStarted, deliveryEnded)
}
mu.Lock()
defer mu.Unlock()
if len(order) != 2 {
t.Fatalf("fetched %v", order)
if len(order) != 2 || order[0] != askCommit || order[1] != checkCommit {
t.Fatalf("fetched %v, want the delivery ask's commit %s first, then the check's %s", order, askCommit, checkCommit)
}
}
@@ -174,7 +174,7 @@ func TestACheckAsksDigestIsTheToolsAnswerInTheCall(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if l := out.(delivery.CheckAnswer).Levels; len(l) != 1 || !strings.Contains(string(l[0].Digest), `"x/a":"d1"`) {
if l := out.(delivery.CheckAnswer).Levels; len(l) != 1 || l[0].Digest.Map()["x/a"] != "d1" {
t.Fatalf("levels = %+v, want the digest's packages", l)
}
}
@@ -1,6 +1,7 @@
package main
import (
"encoding/json"
"os"
"strings"
"testing"
@@ -27,7 +28,7 @@ func TestAFailedTestKeepsItsFatalMessageAfterItsLogLines(t *testing.T) {
}
}
if want := "3 subscription(s) derive to a subject no emitter owns:"; f.Cause != want && !strings.HasSuffix(f.Cause, want) {
t.Errorf("1901/AC2: cause = %q, want the t.Fatalf line, not the t.Logf line", f.Cause)
t.Errorf("1901/AC1: cause = %q, want the t.Fatalf line, not the t.Logf line", f.Cause)
}
}
@@ -38,11 +39,74 @@ func TestAFailureLongerThanTheBoundKeepsItsLastLines(t *testing.T) {
lines = append(lines, " x_test.go:1: log line "+strings.Repeat("y", 40))
}
lines = append(lines, " x_test.go:9: the real cause", "--- FAIL: TestX (0.00s)")
out := clip(lines)
out := clip(lines, "")
if len(out) > outputBytes+100 {
t.Errorf("1901/AC1: kept %d bytes, bound is %d", len(out), outputBytes)
}
if !strings.HasSuffix(out, "the real cause\n--- FAIL: TestX (0.00s)") || !strings.HasPrefix(out, "... (") {
if !strings.HasSuffix(out, "the real cause\n--- FAIL: TestX (0.00s)") || !strings.Contains(out, "\n... (") {
t.Errorf("1901/AC1: out = %q...%q", out[:40], out[len(out)-60:])
}
}
// An older checker, which knows no "cause" field, still reads the new build module's answer, and the output it gets
// holds at least what the older build module kept (its first lines, for an output shorter than the bound).
func TestAnOlderCheckerReadsTheNewFailureAndGetsAtLeastWhatItKept(t *testing.T) { // 1901/AC3
raw, err := os.ReadFile("testdata/captured/go-test-json-log-then-fatal.jsonl")
if err != nil {
t.Fatal(err)
}
res := &TestResult{}
rd := newReader(res, ".")
rd.read(string(raw))
rd.finish()
b, err := json.Marshal(res.Failed)
if err != nil {
t.Fatal(err)
}
var old []struct {
Package string `json:"package"`
Test string `json:"test,omitempty"`
Output string `json:"output,omitempty"`
}
if err := json.Unmarshal(b, &old); err != nil || len(old) != 1 {
t.Fatalf("1901/AC3: an older checker refused the answer: %v, %d failures", err, len(old))
}
first := strings.SplitN(old[0].Output, "\n", 2)[0]
if first == "" || !strings.Contains(old[0].Output, "judged against") || !strings.Contains(old[0].Output, "3 subscription(s)") {
t.Errorf("1901/AC3: older checker's output = %q", old[0].Output)
}
}
// A long failure, whose earlier lines are left out, still starts with a real line for an older checker, which keeps only the
// first non-empty line of the output cut to 120 characters, and not with the marker.
func TestALongFailureStillStartsWithARealLineForAnOlderChecker(t *testing.T) { // 1901/AC3
var lines []string
for i := 0; i < 400; i++ {
lines = append(lines, " x_test.go:1: log line "+strings.Repeat("y", 40))
}
lines = append(lines, " x_test.go:9: the real cause", "--- FAIL: TestX (0.00s)")
oldFirst := func(out string) string { // the older checker's rule (judge.go firstLine)
for _, l := range strings.Split(out, "\n") {
if l = strings.TrimSpace(l); l != "" {
if len(l) > 120 {
l = l[:120]
}
return l
}
}
return ""
}
for _, cause := range []string{"the real cause", ""} {
out := clip(lines, cause)
got := oldFirst(out)
if strings.HasPrefix(got, "... (") || got == "" {
t.Errorf("1901/AC3: cause %q: an older checker would read %q", cause, got)
}
if cause != "" && got != cause {
t.Errorf("1901/AC3: older checker reads %q, want the cause", got)
}
if !strings.HasSuffix(out, "--- FAIL: TestX (0.00s)") {
t.Errorf("1901/AC3: the last lines are lost: %q", out[len(out)-40:])
}
}
}
@@ -207,15 +207,17 @@ func (r *reader) event(ev event) {
r.times = append(r.times, TestTime{Package: ev.Package, Test: ev.Test, Elapsed: ev.Elapsed})
if ev.Action == "fail" {
r.failedIn[ev.Package] = true
r.res.Failed = append(r.res.Failed, Failure{Package: ev.Package, Test: ev.Test, Output: clip(r.testOut[key]), Cause: r.causes[key]})
r.res.Failed = append(r.res.Failed, Failure{Package: ev.Package, Test: ev.Test, Output: clip(r.testOut[key], r.causes[key]), Cause: r.causes[key]})
}
}
}
}
// clip joins the last lines of a failure's output that fit outputBytes, each cut to outputLineBytes, and says how many
// earlier lines it left out.
func clip(lines []string) string {
// earlier lines it left out. **When it leaves lines out, its first line is still a real one** (issue 1901/AC3): the failure's
// cause, or failing that the first non-empty line of the output, then the marker. An older checker takes only the first
// non-empty line of the output, and a marker there would be less than the older pair kept.
func clip(lines []string, cause string) string {
size, from := 0, len(lines)
for from > 0 {
n := len(cutLine(lines[from-1])) + 1
@@ -227,6 +229,13 @@ func clip(lines []string) string {
}
kept := make([]string, 0, len(lines)-from+1)
if from > 0 {
head := strings.TrimSpace(cause)
for i := 0; head == "" && i < from; i++ {
head = strings.TrimSpace(lines[i])
}
if head != "" {
kept = append(kept, cutLine(head))
}
kept = append(kept, fmt.Sprintf("... (%d earlier lines)", from))
}
for _, l := range lines[from:] {
@@ -256,7 +265,7 @@ func (r *reader) finish() {
for _, name := range r.order {
p := r.pkgs[name]
if !p.Passed && !r.failedIn[name] {
r.res.Failed = append(r.res.Failed, Failure{Package: name, Output: clip(r.pkgOut[name])})
r.res.Failed = append(r.res.Failed, Failure{Package: name, Output: clip(r.pkgOut[name], "")})
}
if !p.NoTests && p.Elapsed > r.res.Elapsed {
r.res.Elapsed = p.Elapsed
@@ -12,6 +12,7 @@ import (
"regexp"
"sort"
"strings"
"sync"
"git.novox.be/novox/mesh-sdk/go/forgekey"
)
@@ -127,3 +128,45 @@ func (m *Module) lint(ctx context.Context, run forgekey.Runner, tree, d string,
}
return failed, nil
}
// lintAll lints each directory, up to m.parallel() at once like the level's tests; the failures join in directory order
// and the error is the first directory's, so the answer does not depend on which ended first (issue 1930).
func (m *Module) lintAll(ctx context.Context, run forgekey.Runner, tree string, dirs []string, env func(d string) []string) ([]Failure, error) {
type answer struct {
failed []Failure
err error
}
answers := make([]answer, len(dirs))
slots := make(chan struct{}, m.parallel())
var wg sync.WaitGroup
var crash any // a panic of a lint goes on in the caller, as a panic of a test run does
var crashMu sync.Mutex
for i, d := range dirs {
wg.Add(1)
slots <- struct{}{}
go func() {
defer wg.Done()
defer func() { <-slots }()
defer func() {
if p := recover(); p != nil {
crashMu.Lock()
crash = p
crashMu.Unlock()
}
}()
answers[i].failed, answers[i].err = m.lint(ctx, run, tree, d, env(d))
}()
}
wg.Wait()
if crash != nil {
panic(crash)
}
var failed []Failure
for _, a := range answers {
if a.err != nil {
return nil, a.err
}
failed = append(failed, a.failed...)
}
return failed, nil
}
@@ -5,7 +5,9 @@ import (
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/forgekey"
)
@@ -90,3 +92,43 @@ func TestAnUnformattedFileFailsTheUnitLevel(t *testing.T) {
}
}
}
// Issue 1930: the lint of many directories runs side by side up to the level's bound, not one after another, and the
// failures still join in directory order.
func TestTheLintOfManyDirectoriesRunsAtOnceUpToTheBound(t *testing.T) {
files := map[string]string{}
var dirs []string
for _, d := range []string{"a", "b", "c", "d", "e", "f"} {
files[d+"/go.mod"] = "module x/" + d + "\n"
files[d+"/x.go"] = "package x\n"
dirs = append(dirs, d)
}
tree := writeTree(t, files)
var mu sync.Mutex
running, most := 0, 0
run := func(ctx context.Context, env []string, name string, args ...string) forgekey.Ran {
mu.Lock()
running++
most = max(most, running)
mu.Unlock()
time.Sleep(30 * time.Millisecond)
mu.Lock()
running--
mu.Unlock()
if strings.HasSuffix(args[1], "b") || strings.HasSuffix(args[1], "e") {
return forgekey.Ran{Status: 1, Stderr: "bad " + filepath.Base(args[1])}
}
return forgekey.Ran{}
}
m := &Module{Parallel: 3}
failed, err := m.lintAll(context.Background(), run, tree, dirs, func(string) []string { return nil })
if err != nil {
t.Fatal(err)
}
if most < 2 || most > 3 {
t.Errorf("%d lint commands ran at once, want 2 or 3 (the bound)", most)
}
if len(failed) != 2 || failed[0].Package != "b" || failed[1].Package != "e" {
t.Errorf("failed = %+v, want b's then e's", failed)
}
}
@@ -0,0 +1,79 @@
package main
import (
"context"
"reflect"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/buildstep"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// 1925/AC1: packagesArg takes the list as the bus's JSON shape ([]any) and as the typed list an ask hands in process
// ([]string), and refuses a bad element in either.
func TestPackagesArgTakesBothShapes(t *testing.T) {
cases := []struct {
name string
in any
want []string
refused bool
}{
{"absent", nil, nil, false},
{"typed", []string{"x/a", "x/b/..."}, []string{"x/a", "x/b/..."}, false},
{"json", []any{"x/a", "x/b/..."}, []string{"x/a", "x/b/..."}, false},
{"typed empty", []string{}, []string{}, false},
{"typed dotdot", []string{"x/../y"}, nil, true},
{"json dotdot", []any{"x/../y"}, nil, true},
{"typed bad", []string{"x a"}, nil, true},
{"json bad", []any{"x a"}, nil, true},
{"json not a string", []any{3}, nil, true},
{"a string", "x/a", nil, true},
}
for _, c := range cases {
got, err := packagesArg(c.in)
if c.refused != (err != nil) {
t.Errorf("%s: err = %v, refused want %v", c.name, err, c.refused)
continue
}
if !c.refused && !reflect.DeepEqual(got, c.want) {
t.Errorf("%s: got %#v, want %#v", c.name, got, c.want)
}
}
}
// 1925/AC2: a check ask through the module's real build_go_test body gets past the packages argument: the digest names
// a package, buildstep.RunAsk hands it in as a typed list, and the step fails only later, on the workspace nobody fetched.
func TestACheckAskReachesTheTestStepPastItsPackages(t *testing.T) {
m := &Module{Node: "ace", Work: t.TempDir()}
var calls []string
steps := fakeSteps(&calls)
for i := range steps {
switch steps[i].Name {
case "build_go_digest":
steps[i].Run = func(ctx context.Context, args map[string]any) (any, error) {
return map[string]any{"level": "unit", "packages": map[string]string{"x/a": "d1"}}, nil
}
case "build_go_test":
for _, real := range m.StepTools() {
if real.Name == "build_go_test" {
steps[i].Run = real.Run
}
}
}
}
a := delivery.StepAsk{ID: buildstep.NewID(), Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-checker",
At: time.Now(), Check: &delivery.CheckJob{Fetch: map[string]any{"repository": "novox/x", "commit": askCommit}, Levels: []string{delivery.LevelUnit}}}
out, err := buildstep.RunAsk(context.Background(), m.askSteps(steps), a)
got := ""
if err != nil {
got = err.Error()
}
if strings.Contains(got, "packages is a list") {
t.Fatalf("the test step refused the ask's packages: %s", got)
}
if !strings.Contains(got, "no workspace") {
t.Fatalf("want the step to fail on the unfetched workspace, past its packages; err=%v out=%+v", err, out)
}
}
+19 -13
View File
@@ -127,17 +127,25 @@ func (m *Module) besideEnv(workspace string) []string {
return []string{"MESH_CHECK_BESIDE=" + dir}
}
// packagesArg reads the packages argument: a list of package patterns, as []any when it came over the bus as JSON and
// as []string when an ask handed it in process (issue 1925).
func packagesArg(v any) ([]string, error) {
if v == nil {
var list []string
switch t := v.(type) {
case nil:
return nil, nil
}
list, ok := v.([]any)
if !ok {
case []string:
list = t
case []any:
for _, e := range t {
s, _ := e.(string)
list = append(list, s)
}
default:
return nil, fmt.Errorf("packages is a list of package patterns")
}
out := []string{}
for _, e := range list {
s, _ := e.(string)
for _, s := range list {
if !packageOK.MatchString(s) || strings.Contains(strings.TrimSuffix(s, "/..."), "..") {
return nil, fmt.Errorf("%q is not a package pattern", s)
}
@@ -201,13 +209,11 @@ func (m *Module) runLevel(ctx context.Context, tree string, o levelOpts) (*TestR
// gofmt and go vet of each directory the level runs come before its tests (ADR 0529 §1; issue 1887 D7-go): a
// directory that fails them fails the level at once, and their time is the result's Lint.
linted := m.clock()
var failed []Failure
for _, d := range todo {
f, err := m.lint(ctx, run, tree, d, append(m.goEnvFor(filepath.Join(tree, d)), o.env...))
if err != nil {
return nil, err
}
failed = append(failed, f...)
failed, err := m.lintAll(ctx, run, tree, todo, func(d string) []string {
return append(m.goEnvFor(filepath.Join(tree, d)), o.env...)
})
if err != nil {
return nil, err
}
res.Lint = m.clock().Sub(linted).Seconds()
if len(failed) > 0 {
+1 -1
View File
@@ -3,7 +3,7 @@ module meshbuildgo
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93
github.com/distribution/distribution/v3 v3.1.2
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93 h1:FAiFY2KBKEoZSH49OZTt7/zRBIQxyl5g/KhYrpggbN4=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/alicebob/miniredis/v2 v2.35.0 h1:QwLphYqCEAo1eu1TqPRN2jgVMPBweeQcR21jeqDCONI=
github.com/alicebob/miniredis/v2 v2.35.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
+1 -1
View File
@@ -3,7 +3,7 @@ module meshbuildimage
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93 h1:FAiFY2KBKEoZSH49OZTt7/zRBIQxyl5g/KhYrpggbN4=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+1 -1
View File
@@ -3,7 +3,7 @@ module meshbuildnode
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93
github.com/distribution/distribution/v3 v3.1.2
github.com/nats-io/nats.go v1.54.0
)
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93 h1:FAiFY2KBKEoZSH49OZTt7/zRBIQxyl5g/KhYrpggbN4=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/alicebob/miniredis/v2 v2.35.0 h1:QwLphYqCEAo1eu1TqPRN2jgVMPBweeQcR21jeqDCONI=
github.com/alicebob/miniredis/v2 v2.35.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
@@ -514,6 +514,11 @@ func TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealB
announced.Title = "Fix the apostrophe (issue 975)"
announced.Body = "The description, with no review line yet."
b.publish(t, "mesh.mod.gitea.event.pull.updated", announced)
rt.until("the pull request's key to hold its head", eventWithin, func() bool {
e, err := b.kvs["pulls"].Get(ctx, "novox.app.7")
var k PullKey
return err == nil && json.Unmarshal(e.Value(), &k) == nil && k.Head == head && k.Trunk == "a1a1a1a1a1a1"
})
b.checkPasses(t, rt, "fix/x", "a1a1a1a1a1a1", judgedPull{"novox/app", 7, head})
// No review names the head, so nothing is asked (ADR 0496, issue 1803): the delivery waits for a review.
@@ -54,6 +54,9 @@ type Holder struct {
seatMu sync.Mutex
seatWhy string
// catchUp asks the next look to weigh again the newest merge of each repository against source.main (SourceCatchUp).
catchUp bool
mu sync.Mutex
deliveries map[string]*Delivery
groups map[string]*Group
@@ -1845,6 +1848,12 @@ func (h *Holder) adopt(w Walk) {
now := h.Now()
d := &Delivery{ID: IDOf(w.Repository, w.Commit), Repository: strings.ToLower(w.Repository), Commit: w.Commit,
Base: w.Branch, MergedAs: w.Commit, Created: now}
// Dated once, by the walk it is adopted from, which follows the merge: a later walk it follows (linkWalk) never redates
// it, so it never looks newer than a merge heard after it (issue 1726, review).
d.MergedAt = w.Created.UTC()
if w.Created.IsZero() {
d.MergedAt = now
}
if old := h.deliveries[d.ID]; old != nil {
if old.State.Final() {
// Never revived: an ended delivery is not the walk's; the walk, held by nothing, is listed by
@@ -0,0 +1,161 @@
package main
import (
"errors"
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
sdk "git.novox.be/novox/mesh-sdk/go/delivery"
)
// Issue 1726 (2026-10-24, mesh-catalog #471 and #472): the checker takes a repository's trunk head only from source.main
// (mesh-delivery modules/mesh-delivery-checker main.go), and the gate moves source.main only on a merge it hears. While
// gitea's watcher was dead, the operator merged #471; the restarted watcher's poll announced its pull.merged while the gate
// was starting, before it could tell whether it held the seat, and the gate acknowledged it unread. The gate then adopted the
// controller's walk of the merge commit, which has no merge time, so every older merge of the repository looked newer and
// source.main was never moved: the checker judged #472 on the trunk before #471, and the gate refused its merge.
const (
i1726Repo = "novox/mesh-catalog"
i1726Head1 = "aaaa000000000000000000000000000000000001"
i1726Merge1 = "dddd000000000000000000000000000000000002"
i1726Merge2 = "bbbb000000000000000000000000000000000003"
)
// 1726c/AC1: a merge into main whose pull.merged the gate never heard, caught up as the controller's walk of its merge
// commit, moves source.main as the forge's word of it would.
func TestAMergeCaughtUpAsAnAdoptedWalkMovesSourceMain(t *testing.T) {
w := newWorld(t)
w.h.PullUpdated(pr(i1726Repo, 470, i1726Head1, "fix/a", "modules/app/a.go"))
w.h.PullMerged(i888MergedAt(i1726Repo, 470, i1726Head1, i1726Merge1, w.now.Add(-time.Hour)))
w.settleAll()
if w.source.main == nil || w.source.main.Heads[i1726Repo] != i1726Merge1 {
t.Fatalf("the heard merge did not move source.main: %+v", w.source.main)
}
// #471's pull.merged is lost; the controller walks its merge commit and the gate adopts the walk.
walk := aWalk("plan-471", i1726Repo, i1726Merge2, false)
walk.Created = w.now.Add(-time.Minute)
walk.State = walkRolling
w.ctl.put(walk)
if err := w.h.Reconcile(); err != nil {
t.Fatal(err)
}
w.settleAll()
if got := w.source.main.Heads[i1726Repo]; got != i1726Merge2 {
t.Fatalf("source.main holds %s at %s after the walk of %s was adopted; wanted the adopted merge", i1726Repo, got, i1726Merge2)
}
}
// 1726c/AC2: a merge already marked in the source without being written (as the gate before this fix left #471) is written
// when the gate next holds the seat; one older than the head source.main holds never moves it back.
func TestHoldingTheSeatWritesTheNewestMergeOfEachRepositoryNotYetInSourceMain(t *testing.T) {
w := newWorld(t)
w.h.PullUpdated(pr(i1726Repo, 470, i1726Head1, "fix/a", "modules/app/a.go"))
w.h.PullMerged(i888MergedAt(i1726Repo, 470, i1726Head1, i1726Merge1, w.now.Add(-time.Hour)))
w.settleAll()
walk := aWalk("plan-471", i1726Repo, i1726Merge2, false)
walk.Created = w.now.Add(-time.Minute)
walk.State = walkRolling
w.ctl.put(walk)
if err := w.h.Reconcile(); err != nil {
t.Fatal(err)
}
w.settleAll()
// What the gate before this fix left: the adopted merge marked in the source, source.main still at the merge before.
if d := w.delivery(IDOf(i1726Repo, i1726Merge2)); !d.InSource {
t.Fatalf("the adopted merge is not marked in the source: %+v", d)
}
w.source.main = &sdk.Source{V: sdk.Version, Group: "before", Heads: map[string]string{i1726Repo: i1726Merge1, "novox/other": "b0"}}
w.settleAll()
if got := w.source.main.Heads[i1726Repo]; got != i1726Merge1 {
t.Fatalf("source.main moved to %s before the gate held the seat again", got)
}
w.h.SourceCatchUp()
w.settleAll()
if got := w.source.main.Heads[i1726Repo]; got != i1726Merge2 || w.source.main.Heads["novox/other"] != "b0" {
t.Fatalf("source.main is %+v after the catch-up; wanted %s at %s and the other kept", w.source.main.Heads, i1726Repo, i1726Merge2)
}
n := len(w.source.moves)
w.h.SourceCatchUp()
w.settleAll()
if len(w.source.moves) != n {
t.Fatalf("a second catch-up moved source.main again: %d moves, then %d", n, len(w.source.moves))
}
o := newWorld(t)
o.source.main = &sdk.Source{V: sdk.Version, Group: "newer", Heads: map[string]string{i1726Repo: i1726Merge2}}
o.h.PullUpdated(pr(i1726Repo, 471, "cccc000000000000000000000000000000000001", "fix/b", "modules/app/b.go"))
o.h.PullMerged(i888MergedAt(i1726Repo, 471, "cccc000000000000000000000000000000000001", i1726Merge2, o.now.Add(-time.Minute)))
o.h.PullMerged(i888MergedAt(i1726Repo, 470, i1726Head1, i1726Merge1, o.now.Add(-time.Hour)))
o.settleAll()
o.h.SourceCatchUp()
o.settleAll()
if got := o.source.main.Heads[i1726Repo]; got != i1726Merge2 {
t.Fatalf("the catch-up moved source.main back to %s", got)
}
}
// seatUnknown is a seat whose occupants cannot be read yet.
type seatUnknown struct{}
func (seatUnknown) Occupants() ([]string, error) {
return nil, errors.New("the watches have not delivered yet")
}
// seatTaken is a seat today's module still stands on.
type seatTaken struct{}
func (seatTaken) Occupants() ([]string, error) { return []string{"assigned on elsewhere"}, nil }
// 1726c/AC3: an event that reaches the gate before it can tell whether it holds the seat is offered again, never
// acknowledged unread; one that reaches a gate whose seat another holds is acknowledged and left, as before.
func TestAnEventBeforeTheGateKnowsWhetherItHoldsTheSeatIsOfferedAgain(t *testing.T) {
w := newWorld(t)
w.h.Seat = seatUnknown{}
_ = w.h.SeatChanged()
ev := stdio.Envelope{Key: "gitea.pull.merged", Body: mustJSON(i888MergedAt(i1726Repo, 471, i1726Head1, i1726Merge2, w.now))}
if err := handle(w.h, ev); err == nil {
t.Fatal("an event heard before the gate could tell whether it holds the seat was taken: acknowledged unread")
}
w.h.Seat = seatTaken{}
_ = w.h.SeatChanged()
if err := handle(w.h, ev); err != nil {
t.Fatalf("an event heard while today's module holds the seat was offered again: %v", err)
}
}
// 1726c/AC4 (review round 1): an adopted merge that later follows a newer walk keeps the date of the walk it was adopted
// from, so a merge heard from the forge after it stays in source.main through a catch-up.
func TestAnAdoptedMergeFollowingANewerWalkNeverMovesSourceMainBack(t *testing.T) {
w := newWorld(t)
walk := aWalk("plan-471", i1726Repo, i1726Merge2, false)
walk.Created = w.now.Add(-10 * time.Minute)
walk.State = walkRolling
w.ctl.put(walk)
if err := w.h.Reconcile(); err != nil {
t.Fatal(err)
}
w.settleAll()
if got := w.source.main.Heads[i1726Repo]; got != i1726Merge2 {
t.Fatalf("the adopted merge did not move source.main: %s", got)
}
const later = "eeee000000000000000000000000000000000004"
w.h.PullUpdated(pr(i1726Repo, 473, "ffff000000000000000000000000000000000001", "fix/c", "modules/app/c.go"))
w.h.PullMerged(i888MergedAt(i1726Repo, 473, "ffff000000000000000000000000000000000001", later, w.now.Add(-5*time.Minute)))
w.settleAll()
if got := w.source.main.Heads[i1726Repo]; got != later {
t.Fatalf("the later forge merge did not move source.main: %s", got)
}
again := aWalk("plan-471-again", i1726Repo, i1726Merge2, false)
again.Created = w.now.Add(-time.Minute)
again.State = walkRolling
w.h.mu.Lock()
w.h.linkWalk(w.h.deliveries[IDOf(i1726Repo, i1726Merge2)], again, i1726Merge2)
w.h.mu.Unlock()
w.h.SourceCatchUp()
w.settleAll()
if got := w.source.main.Heads[i1726Repo]; got != later {
t.Fatalf("the catch-up moved source.main back to the adopted merge %s after it followed a newer walk", got)
}
}
@@ -147,6 +147,7 @@ func run(h *Holder, l *listening, occupancy *busOccupancy) {
func readOnHolding(h *Holder) {
h.wantForge()
h.hearAsks()
h.SourceCatchUp() // a merge missed or marked unwritten meanwhile reaches source.main, and so the checker (issue 1726)
h.Poke()
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
err := h.Reconcile()
@@ -162,6 +163,11 @@ func readOnHolding(h *Holder) {
// handle takes one event, then asks for a look at every step the event makes possible (Work, React). An event that cannot be read is
// said and acknowledged: read again it would fail again, and the controller's walks are read back by comparison anyway.
func handle(h *Holder, e stdio.Envelope) error {
if h.Undecided() {
// Not known yet whether this gate holds the seat (starting, or carrying today's state): offered again in a few
// seconds, so an event heard at a restart is not lost (issue 1726).
return fmt.Errorf("%s: the gate cannot tell yet whether it holds the seat mesh-delivery; offered again", e.Key)
}
if !h.Holding() {
return nil // not the holder of the seat: the event is acknowledged and taken by no one (seat.go)
}
@@ -433,10 +433,20 @@ func gateMerges(d *Delivery) bool {
// and the checker judges on the trunk source.main names, so source.main moves too. A merge replayed as history, or older than
// a merge of its repository already heard, is never written: source.main never moves backwards. Under the lock; the writing
// happens outside it.
//
// A merge caught up after a gap (issue 1726) is settled the same way: the controller's walk of a merge commit whose
// pull.merged the gate never heard is adopted with the walk's time as its merge's (adopt, mergedWhen), and once each time the
// gate comes to hold the seat (SourceCatchUp) the newest merge of each repository it holds is weighed again though it is
// marked in the source, so a merge marked there and never written reaches source.main, and with it the checker.
func (h *Holder) settleForgeMerges() []func() {
var out []func()
again := map[*Delivery]bool{}
if h.catchUp {
h.catchUp = false
again = h.newestInSource()
}
for _, d := range h.deliveries {
if d.MergedAs == "" || d.InSource || gateMerges(d) {
if d.MergedAs == "" || (d.InSource && !again[d]) || gateMerges(d) {
continue
}
if d.Base != MergeBranch || d.History != "" || h.newerMergeOf(d) {
@@ -444,7 +454,7 @@ func (h *Holder) settleForgeMerges() []func() {
h.keep(d)
continue
}
id, repository, base, commit, at, group := d.ID, d.Repository, d.Base, d.MergedAs, d.MergedAt, firstOf(d.Branch, d.Repository)
id, repository, base, commit, at, group := d.ID, d.Repository, d.Base, d.MergedAs, mergedWhen(d), firstOf(d.Branch, d.Repository)
out = append(out, func() {
err := h.Source.Move(func(prev *sdk.Source) (sdk.Source, bool) {
next := sdk.Source{Group: group, Heads: map[string]string{repository: commit}}
@@ -485,7 +495,7 @@ func (h *Holder) settleForgeMerges() []func() {
// (Ready.MergeCommit, Ready.Merged), which comes first for a merge the gate handed. Empty when it is not merged.
func mergeOf(d *Delivery) (string, time.Time) {
if d.MergedAs != "" {
return d.MergedAs, d.MergedAt
return d.MergedAs, mergedWhen(d)
}
if d.Ready != nil && d.Ready.Merged != nil && d.Ready.MergeCommit != "" {
return d.Ready.MergeCommit, *d.Ready.Merged
@@ -497,7 +507,7 @@ func mergeOf(d *Delivery) (string, time.Time) {
func (h *Holder) newerMergeOf(d *Delivery) bool {
for _, o := range h.deliveries {
if commit, at := mergeOf(o); o != d && o.Repository == d.Repository && o.Base == d.Base && commit != "" &&
!sameCommit(commit, d.MergedAs) && at.After(d.MergedAt) {
!sameCommit(commit, d.MergedAs) && at.After(mergedWhen(d)) {
return true
}
}
@@ -518,3 +528,44 @@ func (h *Holder) olderHead(repository, base, head string, at time.Time) bool {
}
return at.After(h.Now().Add(-keptFinal))
}
// mergedWhen is when a merged delivery's merge landed: its merge time (the forge's, or for a merge adopted from the
// controller's walk the walk's creation, set once in adopt), else when the gate made the delivery. A zero time made every
// other merge of the repository look newer, so an adopted merge never moved source.main (issue 1726).
func mergedWhen(d *Delivery) time.Time {
if !d.MergedAt.IsZero() {
return d.MergedAt
}
return d.Created
}
// SourceCatchUp asks the next look to weigh again, once, the newest merge into the trunk of each repository though it is
// marked in the source (issue 1726): what the gate missed while it did not hold the seat, or marked without writing, reaches
// source.main. The write never moves a repository back (olderHead), and one already there is not written again.
func (h *Holder) SourceCatchUp() {
h.mu.Lock()
h.catchUp = true
h.mu.Unlock()
h.Poke()
}
// newestInSource is, for each repository, its newest merge into the trunk heard from the forge or caught up from a walk,
// marked in the source and not history. Under the lock.
func (h *Holder) newestInSource() map[*Delivery]bool {
newest := map[string]*Delivery{}
for _, d := range h.deliveries {
if d.MergedAs == "" || !d.InSource || gateMerges(d) || d.Base != MergeBranch || d.History != "" {
continue
}
if o := newest[d.Repository]; o == nil || mergedWhen(d).After(mergedWhen(o)) {
newest[d.Repository] = d
}
}
out := map[*Delivery]bool{}
for _, d := range newest {
if !h.newerMergeOf(d) {
out[d] = true
}
}
return out
}
@@ -51,7 +51,38 @@ func (h *Holder) KeepPull(p PullEvent) {
if err != nil {
h.Logf("[mesh-delivery] the key of %s/%s#%d could not be written: %v", p.Owner, p.Repo, p.Number, err)
h.again("a pull request's key that could not be written")
return
}
// The head's mesh/check may have been heard before this key held the head: the two events are taken on goroutines of
// their own, and KeepCheckVerdict then found no head to write a verdict to. The verdict the delivery holds for this head
// and trunk is written now; read after the write above landed, it cannot miss one taken before that write.
verdict := h.checkVerdictOf(p.repository(), p.HeadSHA, p.BaseSHA)
if verdict == "" {
return
}
err = h.Pulls.Change(pullKeyName(p.Owner, p.Repo, p.Number), func(k PullKey) (PullKey, bool) {
if !sameCommit(k.Head, p.HeadSHA) || !sameCommit(k.Trunk, p.BaseSHA) || k.Verdict != "" {
return k, false
}
k.Verdict = verdict
return k, true
})
if err != nil {
h.Logf("[mesh-delivery] the verdict in the key of %s/%s#%d could not be written: %v", p.Owner, p.Repo, p.Number, err)
h.again("a pull request's key that could not be written")
}
}
// checkVerdictOf is the verdict of the mesh/check the delivery of a head holds, when it judged this head on this trunk
// commit; empty otherwise. Outside the lock.
func (h *Holder) checkVerdictOf(repository, head, trunk string) string {
h.mu.Lock()
defer h.mu.Unlock()
d := h.deliveries[IDOf(strings.ToLower(repository), head)]
if d == nil || d.Check == nil || d.Check.Status != ctxCheck || !sameCommit(d.Check.Head, head) || !sameCommit(d.Check.Trunk, trunk) {
return ""
}
return d.Check.Gate
}
// KeepCheckVerdict writes the verdict of a head's mesh/check into the key of its pull request, when the key holds that head
@@ -145,3 +145,21 @@ func TestTheSameMergeHeardAgainAfterALaterMergeMovesNoTrunkBack(t *testing.T) {
t.Fatalf("the trunk moved from %q to %q on a merge heard again; wanted it to stay m9m9m9m9m9m9", before.TrunkNow, after.TrunkNow)
}
}
// The forge's mesh/check event handled wholly between the announcement's PullUpdated and its KeepPull (the two events are
// taken on goroutines of their own): the verdict must still reach the pull request's key.
func TestAVerdictHeardBeforeTheKeyIsWrittenReachesTheKey(t *testing.T) {
w := newWorld(t)
pulls := newFakePulls()
w.h.Pulls = pulls
p := announcedAt(readyHead, "a1a1a1a1a1a1")
w.h.PullUpdated(p)
s := ForgeStatusEvent{Owner: "novox", Repo: "app", SHA: readyHead, Context: ctxCheck, State: "success",
Description: "pass " + readyHead[:12] + " on a1a1a1a1a1a1 | unit 3 run"}
w.h.CommitStatus(s)
w.h.KeepCheckVerdict(s)
w.h.KeepPull(p)
if got := pulls.get(appKey); got.Head != readyHead || got.Verdict != "pass" {
t.Fatalf("the key holds %+v, want the head with the verdict pass", got)
}
}
@@ -105,6 +105,11 @@ func (h *Holder) SeatChanged() error {
// Holding is whether the gate acts. A holder with no Seat to ask always holds (the tests'; main sets one).
func (h *Holder) Holding() bool { return h.Seat == nil || h.seatHeld.Load() }
// Undecided is whether the gate cannot tell yet whether it holds the seat: it does not hold it, and no reason it waits was
// found (the watches have not delivered, or today's state is being carried). An event heard then is offered again, never
// acknowledged unread (issue 1726: a merge announced at a restart was lost so, and source.main never moved).
func (h *Holder) Undecided() bool { return !h.Holding() && h.SeatWhy() == "" }
// setSeatWhy keeps why the gate waits and says whether it changed.
func (h *Holder) setSeatWhy(why string) bool {
h.seatMu.Lock()
@@ -90,7 +90,9 @@ func TestAcceptancePublishedMailBecomesAnEntryNeverKeepingItsBody(t *testing.T)
}
r.stand.Judge = judgeEveryMailAs("neither", "newsletters", "", "")
r.stand.MessengerCheck = func(map[string]any) (any, error) { return map[string]any{"allowed": true}, nil }
r.stand.Tell = func(map[string]any) (any, error) { return map[string]any{"sent_on": ""}, nil }
r.stand.Tell = func(map[string]any) (any, error) {
return map[string]any{"sent_on": nil, "dropped": []string{"telegram"}}, nil
}
publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "hi", time.Now())
@@ -160,7 +162,7 @@ func TestAcceptanceOneMatteringMailSendsExactlyOneTellNamingIt(t *testing.T) {
var tells recorded[map[string]any]
r.stand.Tell = func(args map[string]any) (any, error) {
tells.add(args)
return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil
return map[string]any{"sent_on": []string{"telegram"}, "dropped": []string{}}, nil
}
publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "invoice", time.Now())
@@ -208,7 +210,7 @@ func TestAcceptanceSenderNameWithAHostNameGivesTheSafeForm(t *testing.T) {
var tells recorded[map[string]any]
r.stand.Tell = func(args map[string]any) (any, error) {
tells.add(args)
return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil
return map[string]any{"sent_on": []string{"telegram"}, "dropped": []string{}}, nil
}
publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "mail.evil.example.com", "address": "x@example.org"}}, "hi", time.Now())
@@ -233,9 +235,9 @@ func TestAcceptanceDroppedTellLeavesMailUntoldAndCountedNextTime(t *testing.T) {
var tellCalls count
r.stand.Tell = func(map[string]any) (any, error) {
if tellCalls.inc() == 1 {
return map[string]any{"sent_on": ""}, nil // dropped: the operator channel's own cap
return map[string]any{"sent_on": nil, "dropped": []string{"telegram"}}, nil // dropped: the operator channel's own cap
}
return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil
return map[string]any{"sent_on": []string{"telegram"}, "dropped": []string{}}, nil
}
publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "hi", time.Now())
@@ -269,3 +271,38 @@ func TestAcceptanceDroppedTellLeavesMailUntoldAndCountedNextTime(t *testing.T) {
})
assertNeverSendsMovesFlagsOrDeletes(t, r)
}
// 1926/AC4: a second batch over an unchanged set of mails that matter (its own mail does not matter) sends no
// second signal — the first, answered as the real operator-channel answers it (sent_on a list), marked the set
// told.
func TestAcceptanceRepeatedBatchesOverAnUnchangedSetSendOneSignal(t *testing.T) {
r := startReader(t)
r.stand.MailAccounts = oneAccountAnswer("bob")
r.stand.MailRead = func(map[string]any) (any, error) { return map[string]any{"text": "hi"}, nil }
var judged count
matters := judgeEveryMailAs("matters", "", "", "")
neither := judgeEveryMailAs("neither", "", "", "")
r.stand.Judge = func(args map[string]any) (any, error) {
if judged.inc() == 1 {
return matters(args)
}
return neither(args)
}
r.stand.MessengerCheck = func(map[string]any) (any, error) { return map[string]any{"allowed": true}, nil }
var tells count
r.stand.Tell = func(map[string]any) (any, error) {
tells.inc()
return map[string]any{"sent_on": []string{"telegram"}, "dropped": []string{}}, nil
}
publishAdded(t, r.nc, "bob", "INBOX", 1, []map[string]any{{"name": "Alice", "address": "alice@example.org"}}, "hi", time.Now())
waitFor(t, 30*time.Second, "the first tell", func() bool { return tells.get() >= 1 })
publishAdded(t, r.nc, "bob", "INBOX", 2, []map[string]any{{"name": "Bob", "address": "bob2@example.org"}}, "second", time.Now())
publishAdded(t, r.nc, "bob", "INBOX", 3, []map[string]any{{"name": "Carol", "address": "carol@example.org"}}, "third", time.Now())
waitFor(t, BatchGap+60*time.Second, "later batches judged", func() bool { return judged.get() >= 2 })
time.Sleep(2 * BatchGap) // a stray repeat of the signal would come after a batch
if got := tells.get(); got != 1 {
t.Fatalf("1926/AC4: tells = %d, want exactly 1 for an unchanged set", got)
}
assertNeverSendsMovesFlagsOrDeletes(t, r)
}
@@ -321,7 +321,7 @@ func RunBatches(ctx context.Context, store Store, readText ReadTextFunc, judge J
if err := RunOneBatch(ctx, store, readText, judge, waiting, now, logf); err != nil {
logf("[mesh-mail-reader] a batch failed: %v", err)
}
if err := SendSignal(ctx, store, check, tell); err != nil {
if err := SendSignal(ctx, store, check, tell, now); err != nil {
logf("[mesh-mail-reader] the mail signal could not be sent: %v", err)
}
}
@@ -58,9 +58,9 @@ func TestAcceptanceRefusedJudgementIsUnhealthyThenCatchesUpInPublishedOrder(t *t
var tellCalls count
r.stand.Tell = func(map[string]any) (any, error) {
if tellCalls.inc() == 1 {
return map[string]any{"sent_on": ""}, nil // dropped: the operator channel's own cap (1522/AC2)
return map[string]any{"sent_on": nil, "dropped": []string{"telegram"}}, nil // dropped: the operator channel's own cap (1522/AC2)
}
return map[string]any{"sent_on": time.Now().UTC().Format(time.RFC3339)}, nil
return map[string]any{"sent_on": []string{"telegram"}, "dropped": []string{}}, nil
}
var mu sync.Mutex
@@ -8,6 +8,8 @@ import (
"context"
"encoding/json"
"fmt"
"strings"
"time"
)
// MessengerCheck is one seat:messenger.messenger_check ask (ADR 0461 §4): whether text is safe to say on an
@@ -44,15 +46,75 @@ func NewTell(ask Ask) Tell {
return "", err
}
var out struct {
SentOn string `json:"sent_on"`
SentOn json.RawMessage `json:"sent_on"`
}
if err := json.Unmarshal(raw, &out); err != nil {
return "", fmt.Errorf("operator-channel.tell's answer could not be read: %.200s", raw)
}
return out.SentOn, nil
sentOn, err := readSentOn(out.SentOn)
if err != nil {
return "", fmt.Errorf("operator-channel.tell's answer could not be read: %.200s", raw)
}
return sentOn, nil
}
}
// readSentOn reads tell's sent_on: messenger answers the list of channels that said it (null or empty when
// every channel dropped it), never one string (issue 1926: reading it as a string failed every real answer, so
// no mail was ever marked told and the same signal went out after every batch). A plain string is still
// accepted. The channels are joined with a comma; empty means nothing said it.
func readSentOn(raw json.RawMessage) (string, error) {
if len(raw) == 0 || string(raw) == "null" {
return "", nil
}
var list []string
if err := json.Unmarshal(raw, &list); err == nil {
return strings.Join(list, ","), nil
}
var one string
if err := json.Unmarshal(raw, &one); err != nil {
return "", err
}
return one, nil
}
// signalEntries is what the signal counts out of the entries marked matters and not yet told (issue 1926):
// a mail whose due date lies before today is left out (AC2: it no longer matters, the mail page still holds
// it); the same mail seen twice (one message id in two folders, issue 1849) and the same notice in two
// languages (the same sender address, mail day and due date) count once (AC3). The entries left out stay
// as they were; the ones that count are returned in the order given.
func signalEntries(entries []Entry, now time.Time) []Entry {
today := time.Date(now.UTC().Year(), now.UTC().Month(), now.UTC().Day(), 0, 0, 0, 0, time.UTC)
seen := map[string]bool{}
var out []Entry
for _, e := range entries {
if !e.DueDate.IsZero() && e.DueDate.UTC().Before(today) {
continue
}
var keys []string
if e.MessageID != "" {
keys = append(keys, "id\x00"+e.Account+"\x00"+e.MessageID)
}
if !e.DueDate.IsZero() && e.SenderAddress != "" {
keys = append(keys, "notice\x00"+e.Account+"\x00"+strings.ToLower(e.SenderAddress)+"\x00"+
e.Date.UTC().Format("2006-01-02")+"\x00"+e.DueDate.UTC().Format("2006-01-02"))
}
dup := false
for _, k := range keys {
if seen[k] {
dup = true
}
}
for _, k := range keys {
seen[k] = true
}
if !dup {
out = append(out, e)
}
}
return out
}
// SendSignal is 1522: every entry marked matters and not yet told — including one carried over from an
// earlier batch that went untold (1429/AC4) — is gathered; its words are built (signal.go) and checked word
// by word with check; any one refused falls back to the safe form (1522/AC1). Nothing is sent, and nothing
@@ -61,7 +123,7 @@ func NewTell(ask Ask) Tell {
// own last outcome. The tell sent, every one of those entries is marked told only once it answers sent_on
// (1522/AC2); a tell that answers sent_on empty leaves them exactly as they were, for the next signal to
// count (1522/AC2, AC3).
func SendSignal(ctx context.Context, store Store, check MessengerCheck, tell Tell) error {
func SendSignal(ctx context.Context, store Store, check MessengerCheck, tell Tell, now func() time.Time) error {
health, err := store.ReaderHealth(ctx)
if err != nil {
return err
@@ -73,17 +135,23 @@ func SendSignal(ctx context.Context, store Store, check MessengerCheck, tell Tel
if err != nil {
return err
}
var toTell []Entry
var mails []SignalMail
var untold []Entry
for _, e := range entries {
if e.Mark == MarkMatters && !e.Told {
toTell = append(toTell, e)
mails = append(mails, SignalMail{SenderName: e.SenderName, Summary: e.Summary, Date: e.Date, DueDate: e.DueDate})
untold = append(untold, e)
}
}
// What counts (signalEntries) is what is said; every untold entry is marked told once it is said, so an
// unchanged set is never said twice (issue 1926/AC1) and one that only holds past-due or duplicate mails
// says nothing.
toTell := signalEntries(untold, now())
if len(toTell) == 0 {
return nil
}
var mails []SignalMail
for _, e := range toTell {
mails = append(mails, SignalMail{SenderName: e.SenderName, Summary: e.Summary, Date: e.Date, DueDate: e.DueDate})
}
words := BuildSignal(mails)
allowed := true
@@ -108,7 +176,7 @@ func SendSignal(ctx context.Context, store Store, check MessengerCheck, tell Tel
if sentOn == "" {
return nil
}
for _, e := range toTell {
for _, e := range untold {
e.Told = true
if err := store.UpdateEntry(ctx, e); err != nil {
return err
@@ -2,6 +2,7 @@ package main
import (
"context"
"encoding/json"
"testing"
"time"
)
@@ -38,7 +39,7 @@ func TestSendSignalChecksWordsThenTellsWithSubjectMail(t *testing.T) {
check, asked := recordingCheck(true)
tell, headlines, subjects := recordingTell("2026-10-21T10:00:00Z")
if err := SendSignal(ctx, store, check, tell); err != nil {
if err := SendSignal(ctx, store, check, tell, time.Now); err != nil {
t.Fatal(err)
}
if len(*asked) == 0 || (*asked)[0] != "1 mail matters" {
@@ -64,7 +65,7 @@ func TestSendSignalFallsBackToTheSafeFormWhenRefused(t *testing.T) {
check, _ := recordingCheck(false)
tell, headlines, _ := recordingTell("sent")
if err := SendSignal(ctx, store, check, tell); err != nil {
if err := SendSignal(ctx, store, check, tell, time.Now); err != nil {
t.Fatal(err)
}
if len(*headlines) != 1 {
@@ -81,7 +82,7 @@ func TestSendSignalSentOnEmptyLeavesMailsUntold(t *testing.T) {
check, _ := recordingCheck(true)
tellDropped, calls, _ := recordingTell("") // every channel dropped it
if err := SendSignal(ctx, store, check, tellDropped); err != nil {
if err := SendSignal(ctx, store, check, tellDropped, time.Now); err != nil {
t.Fatal(err)
}
entry, _, _ := store.Entry(ctx, id)
@@ -94,7 +95,7 @@ func TestSendSignalSentOnEmptyLeavesMailsUntold(t *testing.T) {
// The next signal counts the still-untold mail (1429/AC4): SendSignal asks about it again.
tellSent, headlines, _ := recordingTell("sent-now")
if err := SendSignal(ctx, store, check, tellSent); err != nil {
if err := SendSignal(ctx, store, check, tellSent, time.Now); err != nil {
t.Fatal(err)
}
if len(*headlines) != 1 {
@@ -114,7 +115,7 @@ func TestSendSignalSendsNothingForScamWarningsAlone(t *testing.T) {
check, asked := recordingCheck(true)
tell, headlines, _ := recordingTell("sent")
if err := SendSignal(ctx, store, check, tell); err != nil {
if err := SendSignal(ctx, store, check, tell, time.Now); err != nil {
t.Fatal(err)
}
if len(*asked) != 0 || len(*headlines) != 0 {
@@ -132,7 +133,7 @@ func TestSendSignalSendsNothingWhileUnhealthy(t *testing.T) {
check, asked := recordingCheck(true)
tell, headlines, _ := recordingTell("sent")
if err := SendSignal(ctx, store, check, tell); err != nil {
if err := SendSignal(ctx, store, check, tell, time.Now); err != nil {
t.Fatal(err)
}
if len(*asked) != 0 || len(*headlines) != 0 {
@@ -141,7 +142,7 @@ func TestSendSignalSendsNothingWhileUnhealthy(t *testing.T) {
// After recovery, the first signal counts the mail that went untold while unhealthy.
store.SetReaderHealth(ctx, ReaderHealth{Healthy: true})
if err := SendSignal(ctx, store, check, tell); err != nil {
if err := SendSignal(ctx, store, check, tell, time.Now); err != nil {
t.Fatal(err)
}
if len(*headlines) != 1 {
@@ -166,3 +167,120 @@ func TestTheManifestInvokesTellAndMessengerCheck(t *testing.T) {
}
}
}
var signalNow = func() time.Time { return time.Date(2026, 10, 23, 21, 0, 0, 0, time.UTC) }
// 1926/AC1: the real operator-channel.tell answers sent_on as the list of channels that said it (messenger's
// own shape); read as a string it failed, nothing was marked told and every batch sent the same signal again.
func TestNewTellReadsSentOnAsMessengerAnswersIt(t *testing.T) {
for name, answer := range map[string]string{
"a list": `{"key":"tell.1","sent_on":["telegram"],"dropped":[]}`,
"a string": `{"sent_on":"telegram"}`,
} {
tell := NewTell(func(string, map[string]any) (json.RawMessage, error) { return json.RawMessage(answer), nil })
sentOn, err := tell("h", nil, "mail")
if err != nil || sentOn == "" {
t.Fatalf("1926/AC1: %s: sentOn = %q, err = %v, want a channel", name, sentOn, err)
}
}
for name, answer := range map[string]string{
"null": `{"sent_on":null,"dropped":["telegram"]}`,
"empty list": `{"sent_on":[],"dropped":["telegram"]}`,
"absent": `{"dropped":["telegram"]}`,
} {
tell := NewTell(func(string, map[string]any) (json.RawMessage, error) { return json.RawMessage(answer), nil })
sentOn, err := tell("h", nil, "mail")
if err != nil || sentOn != "" {
t.Fatalf("1926/AC1: %s: sentOn = %q, err = %v, want empty and no error", name, sentOn, err)
}
}
}
// 1926/AC1: a set already told is not sent again; a mail added to it is sent alone.
func TestSendSignalSendsAnUnchangedSetOnce(t *testing.T) {
ctx := context.Background()
store := newMemStore()
for i := 0; i < 3; i++ {
store.AddEntry(ctx, mattersEntry("Alice", "pay the bill"))
}
check, _ := recordingCheck(true)
tell, headlines, _ := recordingTell("telegram")
for i := 0; i < 3; i++ {
if err := SendSignal(ctx, store, check, tell, signalNow); err != nil {
t.Fatal(err)
}
}
if len(*headlines) != 1 || (*headlines)[0] != "3 mails matter" {
t.Fatalf("1926/AC1: tell headlines = %v, want one \"3 mails matter\"", *headlines)
}
store.AddEntry(ctx, mattersEntry("Bob", "new"))
if err := SendSignal(ctx, store, check, tell, signalNow); err != nil {
t.Fatal(err)
}
if len(*headlines) != 2 || (*headlines)[1] != "1 mail matters" {
t.Fatalf("1926/AC1: tell headlines = %v, want the changed set sent again", *headlines)
}
}
// 1926/AC2: a mail whose due date has passed is not counted and does not lead; one due today still leads.
func TestSendSignalLeavesPastDueMailsOut(t *testing.T) {
ctx := context.Background()
store := newMemStore()
past := mattersEntry("bpost", "parcel notice")
past.DueDate = time.Date(2026, 10, 22, 0, 0, 0, 0, time.UTC)
today := mattersEntry("Alice", "pay today")
today.DueDate = time.Date(2026, 10, 23, 0, 0, 0, 0, time.UTC)
store.AddEntry(ctx, past)
store.AddEntry(ctx, today)
check, _ := recordingCheck(true)
var headlines []string
var lines [][]string
tell := func(h string, l []string, _ string) (string, error) {
headlines, lines = append(headlines, h), append(lines, l)
return "telegram", nil
}
if err := SendSignal(ctx, store, check, tell, signalNow); err != nil {
t.Fatal(err)
}
if len(headlines) != 1 || headlines[0] != "1 mail matters" || len(lines[0]) != 1 || lines[0][0] != "Alice — pay today — due 2026-10-23" {
t.Fatalf("1926/AC2: told %v %v, want only Alice's mail due today", headlines, lines)
}
// A set holding only past-due mails says nothing.
store2 := newMemStore()
store2.AddEntry(ctx, past)
headlines = nil
if err := SendSignal(ctx, store2, check, tell, signalNow); err != nil {
t.Fatal(err)
}
if len(headlines) != 0 {
t.Fatalf("1926/AC2: told %v for past-due mails alone, want nothing", headlines)
}
}
// 1926/AC3: the same notice in two languages (same sender, same day, same due date) counts once, and so does
// the same message id in two folders (issue 1849); two different mails of one sender stay two.
func TestSendSignalCountsADuplicateNoticeOnce(t *testing.T) {
ctx := context.Background()
store := newMemStore()
due := time.Date(2026, 10, 30, 0, 0, 0, 0, time.UTC)
day := time.Date(2026, 10, 21, 8, 0, 0, 0, time.UTC)
en := Entry{Account: "bob", MessageID: "<a>", SenderName: "bpost", SenderAddress: "no-reply@bpost.example", Summary: "parcel waiting", Mark: MarkMatters, Date: day, DueDate: due}
nl := en
nl.MessageID, nl.Summary, nl.Date = "<b>", "pakje wacht", day.Add(time.Minute)
inbox := Entry{Account: "bob", MessageID: "<c>", Folder: "INBOX", SenderName: "Carol", SenderAddress: "carol@x.example", Summary: "hello", Mark: MarkMatters, Date: day}
allMail := inbox
allMail.Folder = "All Mail"
other := Entry{Account: "bob", MessageID: "<d>", SenderName: "Carol", SenderAddress: "carol@x.example", Summary: "second", Mark: MarkMatters, Date: day}
for _, e := range []Entry{en, nl, inbox, allMail, other} {
store.AddEntry(ctx, e)
}
check, _ := recordingCheck(true)
tell, headlines, _ := recordingTell("telegram")
if err := SendSignal(ctx, store, check, tell, signalNow); err != nil {
t.Fatal(err)
}
if len(*headlines) != 1 || (*headlines)[0] != "3 mails matter" {
t.Fatalf("1926/AC3: tell headlines = %v, want \"3 mails matter\"", *headlines)
}
}