issue 976: the forge's announcements get a deterministic message id; the announcement files go, the poll's watermark is kept in the module's state

This commit is contained in:
2026-10-17 13:52:47 +02:00
parent 2c8c1543ba
commit add91ff048
6 changed files with 274 additions and 59 deletions
+84
View File
@@ -0,0 +1,84 @@
// **The forge's announcements carry a message id** (novox/hq ADR 0406 §6 and §7, issue 976). A merge, a new head and a
// close are announced by more than one path — the poll, the forge's own event, the merge tool's ask of a check — and
// were kept from being announced twice by files of what had been announced. They are now announced with a message id
// that is the same for the same announcement, whichever path makes it: the runtime makes an event's `x-event-id` header
// the bus's message id (mesh-tools node-tools/internal/bus/bus.go, `nats.MsgId`), and the bus stores a message it has
// seen the id of, within its de-duplication window, once.
//
// The SDK's `emit` takes no id, so this publishes through the broker with the headers `emit` stamps and the id given.
// It returns to `emit` once the SDK takes an id.
//
// The ids: a merge by its merge commit, a head by the head and the trunk commit it was announced against (the check asked
// again against a new trunk commit is another announcement), a close by its head.
import { broker } from "@novox/mesh-sdk/messaging";
import { state as keptState } from "@novox/mesh-sdk/state";
type Said = { owner?: string; repo?: string; number?: number; head_sha?: string; base_sha?: string; merge_commit_sha?: string };
/** The message id of an announcement, or undefined for an event that is not one of these three. */
export function announcementId(type: string, body: unknown): string | undefined {
const b = (body ?? {}) as Said;
const pull = `${b.owner}/${b.repo}#${b.number}`;
switch (type) {
case "pull.merged":
return `merged:${pull}@${b.merge_commit_sha ?? ""}`;
case "pull.updated":
return `updated:${pull}@${b.head_sha ?? ""}:${b.base_sha ?? ""}`;
case "pull.closed":
return `closed:${pull}@${b.head_sha ?? ""}`;
default:
return undefined;
}
}
/** Announce a merge, a head or a close: published with its message id. */
export async function announce(type: string, body: unknown): Promise<void> {
const id = announcementId(type, body);
if (!id) throw new Error(`${type} is not an announcement with a message id`);
const node = process.env.MESH_NODE ?? "unknown";
await broker().publish({
key: type,
node,
body,
headers: {
"x-event-id": id,
"x-source": process.env.MESH_MODULE ?? "unknown",
"x-node": node,
"x-time": new Date().toISOString(),
"content-type": "application/json",
},
});
}
// --- the watermark: the one value of the old files that is not the bus's to remember --------------------------------
/** What the module keeps of how far its poll of merges has looked: a time, in its own state. */
interface Kept {
get(key: string): Promise<{ value: unknown } | null>;
put(key: string, value: unknown): Promise<number>;
}
const WATERMARK = "merges-since";
/** The time the poll of merges last looked from, or "" when none is kept or the state cannot be reached: a merge made
* before it is history, one after it is announced. */
export async function readWatermark(state: Kept = keptState("announced")): Promise<string> {
try {
const entry = await state.get(WATERMARK);
const since = (entry?.value as { since?: string } | undefined)?.since;
return typeof since === "string" ? since : "";
} catch (err) {
console.error(`[gitea] where the poll of merges last looked from could not be read, so none is kept — ${err instanceof Error ? err.message : err}`);
return "";
}
}
/** Keep the time the poll of merges looked from; a state that cannot be reached is said, and the next pass keeps it. */
export async function keepWatermark(since: string, state: Kept = keptState("announced")): Promise<void> {
try {
await state.put(WATERMARK, { since });
} catch (err) {
console.error(`[gitea] where the poll of merges looked from could not be kept — ${err instanceof Error ? err.message : err}`);
}
}
+41 -56
View File
@@ -39,6 +39,7 @@
import { emit, on } from "@novox/mesh-sdk/events";
import { registerModuleTools } from "@novox/mesh-sdk/tools";
import { commitStatusTool } from "./delivery.js";
import { announce, keepWatermark, readWatermark } from "./announce.js";
import { mergeTool } from "./merge.js";
import { GiteaAdmin, GiteaClient, movedSince } from "./client.js";
import { CHECK_CONTEXT, REPO_CHECK_CONTEXT, commentFor, headsToAnnounce, statusesFor, type Announced, type Checked,
@@ -90,32 +91,17 @@ async function pollRepos(client: GiteaClient): Promise<void> {
// everything it sees then predates the watching.
import { existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs";
import { join } from "node:path";
const mergedRecord = process.env.MESH_GITEA_STATE_DIR ? join(process.env.MESH_GITEA_STATE_DIR, "merged-announced.json") : null;
// What has been announced is no longer kept in a file (novox/hq ADR 0406 §6, §7, issue 976): an announcement carries a
// message id (announce.ts), so the same merge announced by the poll and by the forge's event, or again after a restart,
// is stored once by the bus — within its de-duplication window, and past it the consumers' idempotence per merge commit
// holds. This process remembers what it announced meanwhile, so it does not ask the forge for the same files twice.
// The one value kept is where the poll last looked from (`since`), in the module's own state: a merge made while this
// module was away, after it, is announced when it is back; one before it is history. With none kept — the first look
// ever — nothing is announced, because everything it sees then predates the watching.
const announced = new Set<string>();
let primedMerges = false;
// since is the moment the watching began: a merge made before it is history, whatever page of the
// forge's listing it surfaces on. Without it, an old merge past the first page — pushed into view
// as newer pull requests were updated — was announced as if it had just happened, and the mesh
// rebuilt everything built from that repository, once per old merge (2026-09-28).
let since = "";
if (mergedRecord && existsSync(mergedRecord)) {
try {
const kept = JSON.parse(readFileSync(mergedRecord, "utf8")) as string[] | { announced: string[]; since: string };
const list = Array.isArray(kept) ? kept : kept.announced;
for (const sha of list) announced.add(sha);
since = Array.isArray(kept) ? new Date().toISOString() : kept.since;
primedMerges = true;
} catch {
// An unreadable record is treated as no record: prime again rather than re-announce history.
}
}
function keepAnnounced(): void {
if (!mergedRecord) return;
mkdirSync(join(mergedRecord, ".."), { recursive: true });
const tmp = mergedRecord + ".tmp";
writeFileSync(tmp, JSON.stringify({ announced: [...announced].slice(-2000), since }));
renameSync(tmp, mergedRecord);
}
let watermarkRead = false;
// **Only the repositories that moved** (novox/hq issue 250). A merge is a push, and a push moves the
// repository's update time; asking every repository for its pull requests on every tick took longer than the
// tick itself, so ticks piled up and a merge was announced minutes late. A repository unchanged since a
@@ -123,10 +109,14 @@ function keepAnnounced(): void {
let lastLook = "";
const MARGIN_MS = 60_000;
async function pollMerged(client: GiteaClient): Promise<void> {
if (!watermarkRead) {
since = await readWatermark();
primedMerges = since !== "";
watermarkRead = true;
}
const began = new Date().toISOString();
const floor = lastLook && primedMerges ? new Date(Date.parse(lastLook) - MARGIN_MS).toISOString() : "";
const repos = movedSince(await client.listAllRepos(), floor);
let changed = false;
for (const repo of repos) {
const pulls = await client.listPullRequests(repo.owner, repo.name, { state: "closed", sort: "recentupdate", limit: "20" });
for (const pull of pulls) {
@@ -134,13 +124,12 @@ async function pollMerged(client: GiteaClient): Promise<void> {
const closedKey = `closed:${repo.full_name}#${pull.number}`;
if (!pull.merged && pull.state === "closed" && !announced.has(closedKey)) {
if (primedMerges && !!pull.updated_at && !!since && pull.updated_at > since) {
await emit("pull.closed", { owner: repo.owner, repo: repo.name, number: pull.number, title: pull.title,
await announce("pull.closed", { owner: repo.owner, repo: repo.name, number: pull.number, title: pull.title,
head: pull.head, head_sha: pull.head_sha, base: pull.base, html_url: pull.html_url });
console.log(`[gitea] announced ${repo.full_name}#${pull.number} closed unmerged`);
await siblingMoves(client, repo.full_name, pull.number);
}
announced.add(closedKey);
changed = true;
continue;
}
if (!pull.merged || !pull.merge_commit_sha || announced.has(pull.merge_commit_sha)) continue;
@@ -152,11 +141,11 @@ async function pollMerged(client: GiteaClient): Promise<void> {
}
else announced.add(pull.merge_commit_sha);
announced.add(pull.merge_commit_sha);
changed = true;
}
}
if (!primedMerges) since = new Date().toISOString();
if (!primedMerges || changed) keepAnnounced();
// The next look starts from this one's start less the margin; kept, so a restart looks from there.
since = primedMerges ? new Date(Date.parse(began) - MARGIN_MS).toISOString() : began;
await keepWatermark(since);
primedMerges = true;
lastLook = began;
}
@@ -196,7 +185,7 @@ async function announceMerge(client: GiteaClient, repo: RepoOf, pull: PullOf): P
console.error(`[gitea] ${repo.full_name}#${pull.number}: its head's statuses could not be read — ${err instanceof Error ? err.message : err}`);
}
}
await emit("pull.merged", {
await announce("pull.merged", {
owner: repo.owner,
repo: repo.name,
number: pull.number,
@@ -226,32 +215,17 @@ async function announceMerge(client: GiteaClient, repo: RepoOf, pull: PullOf): P
announced.delete(pull.merge_commit_sha);
throw err;
}
keepAnnounced();
return true;
}
// **Every new head of an open pull request is announced, once** (novox/hq to-be 45 §9): the mesh checks it
// against every machine of its facts before it merges. Kept beside the merges' record, so a restart
// announces nothing twice; the first look on a machine with no record announces only what moved in the
// last day, so a forge's whole backlog is not checked at once.
const pullsRecord = process.env.MESH_GITEA_STATE_DIR ? join(process.env.MESH_GITEA_STATE_DIR, "pulls-announced.json") : null;
// **Every new head of an open pull request is announced** (novox/hq to-be 45 §9): the mesh checks it against every
// machine of its facts before it merges. What this process announced is remembered in memory only (novox/hq ADR 0406
// §6, §7, issue 976): the announcement carries a message id (announce.ts), so the poll's, the forge event's and the
// merge tool's are one to the bus. After a start, the first look announces a head only if the forge shows no merge check
// of it yet — a head with one was announced — and only what moved in the last day, so a forge's whole backlog is not
// checked at once and a restart asks no check again against a trunk that moved.
let heads: Announced = {};
let primedPulls = false;
if (pullsRecord && existsSync(pullsRecord)) {
try {
heads = (JSON.parse(readFileSync(pullsRecord, "utf8")) as { heads: Announced }).heads ?? {};
primedPulls = true;
} catch {
// An unreadable record is no record: the first look announces only the last day's.
}
}
function keepHeads(): void {
if (!pullsRecord) return;
mkdirSync(join(pullsRecord, ".."), { recursive: true });
const tmp = pullsRecord + ".tmp";
writeFileSync(tmp, JSON.stringify({ heads }));
renameSync(tmp, pullsRecord);
}
/** **A head, announced once, whichever path saw it first** (novox/hq issue 729): the poll or the forge's event. */
async function announceHead(client: GiteaClient, repo: RepoOf, pull: OpenOf): Promise<boolean> {
const key = `${repo.full_name}#${pull.number}`;
@@ -261,13 +235,12 @@ async function announceHead(client: GiteaClient, repo: RepoOf, pull: OpenOf): Pr
if (!claimHead(heads, key, sha)) return false;
try {
// The announcement the merge tool makes too when it asks a check (trunk.ts, novox/hq ADR 0323 §5).
await emit("pull.updated", await announcementOf(client, repo.owner, repo.name, repo.clone_url ?? "", pull));
await announce("pull.updated", await announcementOf(client, repo.owner, repo.name, repo.clone_url ?? "", pull));
} catch (err) {
if (before === undefined) delete heads[key];
else heads[key] = before;
throw err;
}
keepHeads();
// A sibling's new head puts out of date every verdict that read the sibling at another head (ADR 0382 §5).
await siblingMoves(client, repo.full_name, pull.number);
// Said, so an operator knows the mesh was asked to check it.
@@ -279,6 +252,16 @@ async function announceHead(client: GiteaClient, repo: RepoOf, pull: OpenOf): Pr
}
type OpenOf = Awaited<ReturnType<GiteaClient["listOpenPullsSince"]>>[number];
/** Whether the forge shows a merge check of the head, in any state: only an announced head has one. */
async function alreadyChecked(client: GiteaClient, repo: RepoOf, pull: OpenOf): Promise<boolean> {
try {
const checks = await client.headChecks(repo.owner, repo.name, String(pull.head_sha));
return CHECK_CONTEXT in checks || REPO_CHECK_CONTEXT in checks;
} catch {
return false; // the forge cannot say: announced, and the bus's message id collapses it if it was
}
}
let lastPullLook = "";
// **A sweep now and then looks at every repository** (novox/hq issue 347), whatever moved: the floor is a
// heuristic, and a heuristic that misses is caught by the next sweep rather than never.
@@ -288,7 +271,6 @@ async function pollPulls(client: GiteaClient): Promise<void> {
const began = new Date().toISOString();
const floor = lastPullLook ? new Date(Date.parse(lastPullLook) - MARGIN_MS).toISOString() : "";
const dayAgo = new Date(Date.now() - 24 * 3600_000).toISOString();
let changed = false;
// A pull request opened, reopened or retargeted moves its own update time and not its repository's: the
// pull requests updated since the floor are found by their own (issue 347). A search that fails is a sweep.
let sweep = Date.now() - lastSweep >= SWEEP_MS;
@@ -309,14 +291,17 @@ async function pollPulls(client: GiteaClient): Promise<void> {
for (const pull of headsToAnnounce(repo.full_name, open, heads)) {
const key = `${repo.full_name}#${pull.number}`;
const fresh = primedPulls || (!!pull.updated_at && pull.updated_at > dayAgo);
// The first look after a start remembers nothing: a head the forge already shows a merge check of was announced.
if (fresh && !primedPulls && await alreadyChecked(client, repo, pull)) {
heads[key] = String(pull.head_sha);
continue;
}
if (fresh) {
if (await announceHead(client, repo, pull)) pace.miss(`${key} at ${String(pull.head_sha).slice(0, 8)}`);
} else heads[key] = String(pull.head_sha);
changed = true;
}
if (await trunksMoved(client, repo.owner, repo.name)) changedTrunks = true;
}
if (!primedPulls || changed) keepHeads();
if (changedTrunks) keepTrunks();
changedTrunks = false;
primedPulls = true;
@@ -0,0 +1,82 @@
package busdedup
import (
"context"
"testing"
"time"
"github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
)
// **The same announcement published twice with one message id is stored once** (novox/hq ADR 0406 §6, issue 976). The
// forge module announces a merge, a head and a close with a deterministic `x-event-id` (announce.ts); the runtime makes
// that header the bus's message id (mesh-tools node-tools/internal/bus/bus.go, `nats.MsgId`), and the events stream's
// message de-duplication keeps the second. The publish here is the runtime's own: the headers the module stamps, the id
// as the message id. A real JetStream server is linked into the test and started for it alone (the TypeScript harness
// has none, so this is the Go half of the module's tests).
const subject = "mesh.mod.gitea.event.pull.merged"
func publish(t *testing.T, js nats.JetStreamContext, id string) *nats.PubAck {
t.Helper()
msg := nats.NewMsg(subject)
msg.Header.Set("x-event-id", id)
msg.Header.Set("x-source", "gitea")
msg.Header.Set("content-type", "application/json")
msg.Data = []byte(`{"owner":"novox","repo":"app","number":7,"merge_commit_sha":"mmm222"}`)
ack, err := js.PublishMsg(msg, nats.MsgId(id))
if err != nil {
t.Fatal(err)
}
return ack
}
func TestTheSameAnnouncementPublishedTwiceWithOneMessageIdIsStoredOnce(t *testing.T) {
s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true,
StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
if err != nil {
t.Fatal(err)
}
go s.Start()
if !s.ReadyForConnections(30 * time.Second) {
t.Fatal("a bus for this test did not come up within 30s")
}
t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
nc, err := nats.Connect(s.ClientURL())
if err != nil {
t.Fatal(err)
}
t.Cleanup(nc.Close)
js, err := nc.JetStream()
if err != nil {
t.Fatal(err)
}
// The events stream as the controller makes it: two minutes of message de-duplication.
if _, err := js.AddStream(&nats.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.>"}, Duplicates: 2 * time.Minute}); err != nil {
t.Fatal(err)
}
const id = "merged:novox/app#7@mmm222"
first := publish(t, js, id)
second := publish(t, js, id) // the poll and the forge's event both announced it
if first.Duplicate || !second.Duplicate {
t.Fatalf("the first publish was a duplicate: %v; the second: %v (wanted false, true)", first.Duplicate, second.Duplicate)
}
publish(t, js, "merged:novox/app#8@nnn333") // another merge is another message
jsNew, err := jetstream.New(nc)
if err != nil {
t.Fatal(err)
}
stream, err := jsNew.Stream(context.Background(), "EVENTS")
if err != nil {
t.Fatal(err)
}
info, err := stream.Info(context.Background())
if err != nil {
t.Fatal(err)
}
if info.State.Msgs != 2 {
t.Fatalf("the stream stores %d messages for two announcements published three times; wanted 2", info.State.Msgs)
}
}
+2 -2
View File
@@ -5,7 +5,7 @@
// puts its statuses back to pending through that writer, so where the check runs is said and no second check is asked.
// Each bundle the runtime serves is its own process; from the tools' bundle the pending bypassed the writer's chain.
import { emit } from "@novox/mesh-sdk/events";
import { announce } from "./announce.js";
import type { ToolDefinition } from "@novox/mesh-sdk/tools";
import type { CommitStatus, GiteaClient } from "./client.js";
import { mergeableUnder, newestByContext } from "./checks.js";
@@ -27,7 +27,7 @@ export interface MergeDeps {
}
export function mergeTool(gitea: GiteaClient, given: MergeDeps): ToolDefinition {
const deps = { ...given, emit: given.emit ?? emit, stateDir: given.stateDir ?? process.env.MESH_GITEA_STATE_DIR,
const deps = { ...given, emit: given.emit ?? announce, stateDir: given.stateDir ?? process.env.MESH_GITEA_STATE_DIR,
now: given.now ?? (() => new Date()) };
return {
name: "gitea_merge_pull_request",
+2 -1
View File
@@ -498,7 +498,8 @@
}
],
"state": [
"bound"
"bound",
"announced"
],
"reads": [
"mesh-controller.assignments",
+63
View File
@@ -0,0 +1,63 @@
import assert from "node:assert/strict";
import { test } from "node:test";
// **The forge's announcements carry a message id** (novox/hq ADR 0406 §6, §7, issue 976): the same announcement,
// however many paths make it — the poll, the forge's event, the merge tool's ask — has one id, which the runtime makes
// the bus's message id, so the bus stores it once. This replaces the files of what was announced.
const merged = { owner: "novox", repo: "app", number: 7, head_sha: "aaa111", merge_commit_sha: "mmm222", base: "main" };
const updated = { owner: "novox", repo: "app", number: 7, head_sha: "aaa111", base_sha: "ttt333", base: "main" };
const closed = { owner: "novox", repo: "app", number: 7, head_sha: "aaa111", base: "main" };
test("the same merge, head or close announced twice has the same id, and another has another", async () => {
const { announcementId } = await import("../announce.ts");
assert.equal(announcementId("pull.merged", merged), "merged:novox/app#7@mmm222");
assert.equal(announcementId("pull.merged", { ...merged, title: "worded differently" }), announcementId("pull.merged", merged));
assert.notEqual(announcementId("pull.merged", { ...merged, merge_commit_sha: "other" }), announcementId("pull.merged", merged));
assert.equal(announcementId("pull.updated", updated), "updated:novox/app#7@aaa111:ttt333");
assert.notEqual(announcementId("pull.updated", { ...updated, base_sha: "moved" }), announcementId("pull.updated", updated),
"the check asked again against a new trunk commit is another announcement");
assert.notEqual(announcementId("pull.updated", { ...updated, head_sha: "bbb" }), announcementId("pull.updated", updated));
assert.equal(announcementId("pull.updated", { ...updated, base_sha: undefined }), "updated:novox/app#7@aaa111:");
assert.equal(announcementId("pull.closed", closed), "closed:novox/app#7@aaa111");
});
test("an announcement is published through the broker with its id as the event id, and the provenance emit stamps", async () => {
const { useBroker } = await import("@novox/mesh-sdk/messaging");
const published: any[] = [];
useBroker(() => ({ publish: async (env: any) => { published.push(env); } }) as any);
process.env.MESH_MODULE = "gitea";
process.env.MESH_NODE = "anchor";
const { announce } = await import("../announce.ts");
await announce("pull.merged", merged);
await announce("pull.merged", { ...merged });
assert.equal(published.length, 2);
assert.equal(published[0].key, "pull.merged");
assert.equal(published[0].headers["x-event-id"], "merged:novox/app#7@mmm222");
assert.equal(published[1].headers["x-event-id"], published[0].headers["x-event-id"], "one id for one announcement");
assert.equal(published[0].headers["x-source"], "gitea");
assert.equal(published[0].headers["x-node"], "anchor");
assert.equal(published[0].headers["content-type"], "application/json");
assert.ok(published[0].headers["x-time"]);
assert.equal(published[0].node, "anchor");
assert.deepEqual(published[0].body, merged);
});
test("an announcement of a kind it has no id for is refused, never published with a made-up one", async () => {
const { announce, announcementId } = await import("../announce.ts");
assert.equal(announcementId("repo.created", { owner: "a" }), undefined);
await assert.rejects(() => announce("repo.created", { owner: "a" }), /not an announcement with a message id/);
});
test("the watermark is kept in the module's state, and an unreachable state is no record", async () => {
const { readWatermark, keepWatermark } = await import("../announce.ts");
const kept = new Map<string, unknown>();
const store = { get: async (k: string) => (kept.has(k) ? { key: k, value: kept.get(k), revision: 1 } : null),
put: async (k: string, v: unknown) => { kept.set(k, v); return 1; } };
assert.equal(await readWatermark(store as any), "", "none yet");
await keepWatermark("2026-10-17T12:00:00.000Z", store as any);
assert.equal(await readWatermark(store as any), "2026-10-17T12:00:00.000Z");
const away = { get: async () => { throw new Error("no runtime"); }, put: async () => { throw new Error("no runtime"); } };
assert.equal(await readWatermark(away as any), "", "a state that cannot be reached is no record");
await keepWatermark("2026-10-17T12:01:00.000Z", away as any); // said, not thrown
});