593ecdd · e14cb51 · 76fffa8
@@ -0,0 +1,18 @@
# A failed adapter pull speaks once in the writers' feed (#10135)
When an app's adapter pull turns from ok (or from nothing) to failed, the
forge appends one `adapter.sync_failed` event: "<owner>/<repo> sync failed:
<reason>", linking to the app page. Repeated failures and recoveries append
nothing, so a broken source says it once, not on every pull.
- **Who sees it:** the `writers` audience (`src/core/event_audience.ts`): those
  who may write the repository (`canWrite`). Readers and signed-out visitors
  see nothing.
- **Where:** the home feed on the web, and the terminal's feed screen
  (`src/tui/screens/feed.ts`). Both call `readFeed`, which reads
  `candidatesFor` and its audience filter, so the two doors agree
  (`test/feed-sync-failed.test.ts` checks both).
- **Reason:** stored cut to 200 characters (`PULL_REASON_MAX`), shown cut to
  120 in the line.
- **Timing:** the event is appended after the run's transaction commits, so a
  live viewer never hears of a pull that was rolled back.
@@ -38,6 +38,6 @@
export async function pullNow(
  if (!writer.ok) return writer;
  const record = getAppByRepo(app.db, repo.id);
  if (!record) return err("not_found");
  const pulled = await pullAppNow(app.db, record.id, reachOf(record));
  const pulled = await pullAppNow(app.db, record.id, reachOf(record), Date.now, app.bus);
  return pulled.ok ? ok(pulled.run) : err(pulled.why);
}
@@ -9,7 +9,9 @@
 */
import type { Database } from "bun:sqlite";
import { createPullRun, lastPullCursor, listPullRuns, type PullRun } from "../core/runs.ts";
import { getAppById } from "../core/apps.ts";
import { EventBus } from "../core/events.ts";
import { createPullRun, lastPullCursor, lastPullState, listPullRuns, type PullRun } from "../core/runs.ts";
/** A body over this many bytes is a failed pull, never parsed. */
export const PULL_BODY_MAX = 64 * 1024;
@@ -18,7 +20,7 @@
export const PULL_COUNT_MAX = 1_000_000;
/** The cursor is at most this many characters (code points). */
export const PULL_CURSOR_MAX = 200;
/** A failed pull's reason is cut to this many characters. */
const REASON_MAX = 200;
export const PULL_REASON_MAX = 200;
const FIELDS = ["pulled", "skipped", "cursor", "errors"] as const;
const COUNTS = ["pulled", "skipped", "errors"] as const;
@@ -39,7 +41,7 @@
export type PullAnswer =
/** An answer and how long the request took, as `recordPull` takes it. */
export type PullOutcome = PullAnswer & { readonly durationMs: number };
const fail = (reason: string): PullAnswer => ({ ok: false, reason: reason.slice(0, REASON_MAX) });
const fail = (reason: string): PullAnswer => ({ ok: false, reason: reason.slice(0, PULL_REASON_MAX) });
/**
 * Any control, format, private-use, unassigned or surrogate character, or a
@@ -110,13 +112,26 @@
const durationOf = (ms: number): number => (Number.isFinite(ms) && ms > 0 ? Math
 * the last good cursor stays where the next pull starts. A good answer whose
 * cursor equals the last good pull's is a repeat: it is recorded, with its
 * `pulled` counted as skipped and nothing written, so the same records never
 * count as written twice. The read and the write are one transaction, so two
 * recordings of the same answer cannot both see a fresh cursor.
 * count as written twice. A failed pull after an ok one (or the first) also
 * appends one `adapter.sync_failed` event, after the transaction commits, so
 * a live viewer never hears of a run that rolled back; repeats and recoveries
 * append none. The read and the write are one transaction, so two recordings
 * of the same answer cannot both see a fresh cursor.
 */
      
        junipercrew
        2d ago
        outdated
      
      Re-wrap this comment; the line is far over the width the others keep.
    
export function recordPull(db: Database, appId: string, outcome: PullOutcome): PullRun {
export function recordPull(
  db: Database,
  appId: string,
  outcome: PullOutcome,
  bus: Pick<EventBus, "append"> = new EventBus(db),
): PullRun {
  const durationMs = durationOf(outcome.durationMs);
  return db.transaction((): PullRun => {
  let speak: { repoId: string; reason: string } | undefined;
  const run = db.transaction((): PullRun => {
    if (!outcome.ok) {
      // Only the turn from ok (or from nothing) to failed speaks.
      const record = lastPullState(db, appId) !== "failed" ? getAppById(db, appId) : undefined;
      const reason = outcome.reason.slice(0, PULL_REASON_MAX);
      if (record !== undefined) speak = { repoId: record.repoId, reason };
      return createPullRun(db, {
        appId,
        state: "failed",
@@ -125,7 +140,7 @@
export function recordPull(db: Database, appId: string, outcome: PullOutcome): P
        failed: 0,
        cursor: "",
        durationMs,
        reason: outcome.reason.slice(0, REASON_MAX),
        reason,
        repeat: false,
      });
    }
@@ -142,6 +157,17 @@
export function recordPull(db: Database, appId: string, outcome: PullOutcome): P
      repeat,
    });
  }).immediate();
  if (speak !== undefined) {
    bus.append({
      // The literal stays a literal: the audience-coverage scan
      // (`core/event_audience.ts`) reads event kinds from `append` calls, and
      // a constant would hide this one from it.
      kind: "adapter.sync_failed",
      repoId: speak.repoId,
      payload: { appId, reason: speak.reason },
    });
  }
  return run;
}
/** How many runs `pullRunLines` answers when asked for none in particular, and at most. */
@@ -16,6 +16,7 @@
import type { Database } from "bun:sqlite";
import type { App } from "../app.ts";
import { listApps, type AppRecord } from "../core/apps.ts";
import type { EventBus } from "../core/events.ts";
import { getRepoById } from "../core/repos.ts";
import { lastPullAt, type PullRun } from "../core/runs.ts";
import { resolveOid } from "../git/merge.ts";
@@ -76,6 +77,7 @@
async function pullOnce<Held extends string = never>(
  reach: PullReach,
  now: () => number,
  hold: (adapter: AdapterSpec) => Held | undefined = () => undefined,
  bus?: Pick<EventBus, "append">,
): Promise<PullOnce<Held>> {
  const port = reach.port();
  if (port === undefined) return { ok: false, why: "asleep" };
@@ -88,22 +90,34 @@
async function pullOnce<Held extends string = never>(
    if (held !== undefined) return { ok: false, why: held };
    const started = now();
    const answer = answerOf(await reach.post(port));
    return { ok: true, run: recordPull(db, appId, { ...answer, durationMs: Math.max(0, now() - started) }) };
    return { ok: true, run: recordPull(db, appId, { ...answer, durationMs: Math.max(0, now() - started) }, bus) };
  } finally {
    inFlight.delete(appId);
  }
}
/** Pull one app if it is a running adapter and due: whether a pull was made. */
export async function pullApp(db: Database, appId: string, reach: PullReach, now: () => number = Date.now): Promise<boolean> {
export async function pullApp(
  db: Database,
  appId: string,
  reach: PullReach,
  now: () => number = Date.now,
  bus?: Pick<EventBus, "append">,
): Promise<boolean> {
  const notDue = (adapter: AdapterSpec) =>
    isPullDue(lastPullAt(db, appId), adapter.schedule.every, now()) ? undefined : ("not_due" as const);
  return (await pullOnce(db, appId, reach, now, notDue)).ok;
  return (await pullOnce(db, appId, reach, now, notDue, bus)).ok;
}
/** Pull one app now, whatever its schedule (#10084): the recorded run, or why not. Never wakes it. */
export function pullAppNow(db: Database, appId: string, reach: PullReach, now: () => number = Date.now): Promise<PullOnce> {
  return pullOnce(db, appId, reach, now);
export function pullAppNow(
  db: Database,
  appId: string,
  reach: PullReach,
  now: () => number = Date.now,
  bus?: Pick<EventBus, "append">,
): Promise<PullOnce> {
  return pullOnce(db, appId, reach, now, undefined, bus);
}
/** Who the forge says it is when it pulls: an admin of the app, named for itself. */
@@ -203,7 +217,7 @@
export async function pullDue(
  const pulled: string[] = [];
  for (const record of listApps(app.db)) {
    try {
      if (await pullApp(app.db, record.id, reachOf(record), now)) pulled.push(record.slug);
      if (await pullApp(app.db, record.id, reachOf(record), now, app.bus)) pulled.push(record.slug);
    } catch (error) {
      app.log.warn("could not pull an adapter", { app: record.slug, error: String(error) });
    }
@@ -27,10 +27,10 @@
import type { Database } from "bun:sqlite";
import type { ForgeEvent } from "./events.ts";
import { getAppByRepo } from "./apps.ts";
import { personOwners } from "./orgs.ts";
import { canAdmin, canEnterApp, canSeeOwnerFacts, getRepoById } from "./repos.ts";
import { canAdmin, canEnterApp, canWrite, canSeeOwnerFacts, getRepoById } from "./repos.ts";
import type { User } from "./users.ts";
export type EventAudience = "readers" | "admins" | "actor" | "app";
export type EventAudience = "readers" | "admins" | "actor" | "app" | "writers";
/**
 * Whether this viewer answers for a deleted repository: the forge's admin, who
@@ -243,6 +243,8 @@
export function classifiedAudience(kind: string): EventAudience | undefined {
    kind === "repo.file_released"
  )
    return "admins";
  // A broken adapter sync is for those who may write the repository (`apps/pull.ts`).
  if (kind === "adapter.sync_failed") return "writers";
  // A fleet operation spans repositories; its receipt is its actor's. So is
  // a thought: it belongs to no repository and to nobody but its author.
  if (
@@ -291,6 +293,15 @@
export function audienceFilter(
    admins.set(repoId, yes);
    return yes;
  };
  const writers = new Map<string, boolean>();
  const mayWrite = (repoId: string): boolean => {
    const known = writers.get(repoId);
    if (known !== undefined) return known;
    const repo = getRepoById(db, repoId);
    const yes = repo !== undefined && canWrite(db, repo, viewer);
    writers.set(repoId, yes);
    return yes;
  };
  const entrants = new Map<string, boolean>();
  const mayEnter = (repoId: string): boolean => {
    const known = entrants.get(repoId);
@@ -348,6 +359,8 @@
export function audienceFilter(
        return true;
      case "app":
        return event.repoId === null || mayEnter(event.repoId);
      case "writers":
        return event.repoId !== null && mayWrite(event.repoId);
      case "actor":
        return viewer !== undefined && event.actorId === viewer.id;
      case "admins":
@@ -1588,6 +1588,17 @@
export function listPullRuns(db: Database, appId: string, limit = 50): PullRun[]
    .map(toPullRun);
}
/** The state of an app's newest pull run; undefined before the first. */
export function lastPullState(db: Database, appId: string): "ok" | "failed" | undefined {
  return (
    db
      .query<{ state: "ok" | "failed" }, [string]>(
        `SELECT state FROM pull_run WHERE app_id = ?1 ORDER BY id DESC LIMIT 1`,
      )
      .get(appId)?.state ?? undefined
  );
}
/** When an app's last pull, good or failed, was recorded, in ms since the epoch; undefined before the first. */
export function lastPullAt(db: Database, appId: string): number | undefined {
  const row = db
@@ -161,6 +161,8 @@
export function quietRunsFor(
    .filter((e) => quiet.has(e.id) && repos.get(e.repoId!)?.archivedAt == null)
    .map((e) => ({ repoId: e.repoId!, actorId: e.actorId, id: e.id }));
}
/** How much of a sync failure's stored reason (`PULL_REASON_MAX`, `apps/pull.ts`) a feed line shows. */
const SYNC_REASON_SHOWN = 120;
const clip = (s: string, n = 120): string =>
  s.length > n ? `${s.slice(0, n - 1)}…` : s;
@@ -630,6 +632,17 @@
function fromEvent(
        what: `${/^[aeiou]/.test(kind) ? "an" : "a"} ${kind} run${subject} ${state === "failed" ? "failed" : "finished"}${state === "failed" ? (failure ? `: ${failure}` : "") : summary ? `: ${summary}` : ""}`,
      };
    }
    case "adapter.sync_failed": {
      const reason = clip(typeof p["reason"] === "string" ? p["reason"] : "", SYNC_REASON_SHOWN);
      return {
        ...common,
        kind: "run",
        link: `${base(repo)}/app`,
        urgency: 4,
        said: `${slug(repo)} sync failed: ${reason}`,
        what: `sync failed: ${reason}`,
      };
    }
    case "pull.opened":
      return { ...common, kind: "pull", link: change, what: `opened a pull request${title ? `: ${clip(title)}` : ""}${toldWords}` };
    case "pull.merged":
@@ -0,0 +1,81 @@
/** A pull turning failed speaks once in its writers' feed, and only to them (#10135). */
import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp, type App } from "../src/app.ts";
import { newApp } from "../src/actions/apps.ts";
import { recordPull } from "../src/apps/pull.ts";
import { classifiedAudience } from "../src/core/event_audience.ts";
import { getRepo } from "../src/core/repos.ts";
import { createUser, getUserByName } from "../src/core/users.ts";
import { readFeed } from "../src/feed/feed.ts";
import { candidatesFor } from "../src/feed/items.ts";
import { seedAdmin } from "./helpers/admin.ts";
describe("sync failed feed line", () => {
  let dataDir: string;
  let app: App;
  let appId: string;
  const ok = () => recordPull(app.db, appId, { ok: true, pulled: 1, skipped: 0, errors: 0, cursor: `c${Math.random()}`, durationMs: 1 }, app.bus);
  const fail = (reason: string) => recordPull(app.db, appId, { ok: false, reason, durationMs: 1 }, app.bus);
  const lines = (name: string) => {
    const viewer = getUserByName(app.db, name)!;
    return candidatesFor(app.db, viewer, Date.now(), undefined, false).filter((c) => c.said?.includes("sync failed"));
  };
  beforeAll(async () => {
    dataDir = await mkdtemp(join(tmpdir(), "forge-sync-failed-"));
    app = createApp({ dataDir, port: 0 });
    await seedAdmin(app, { username: "alice", email: "[email protected]", password: "a-good-passphrase" });
    const b = await createUser(app.db, {
      origin: "seed", trusted: true, username: "bob",
      email: "[email protected]", password: "a-good-passphrase",
    });
    if (!b.ok) throw new Error(b.error);
    const made = await newApp(app, getUserByName(app.db, "alice")!, { name: "hello", visibility: "public" } as never);
    if (!made.ok) throw new Error(String(made.error));
    const repo = getRepo(app.db, "alice", "hello")!;
    appId = (app.db.query("SELECT id FROM app WHERE repo_id = ?1").get(repo.id) as { id: string }).id;
  });
  afterAll(async () => {
    app.db.close();
    await rm(dataDir, { recursive: true, force: true });
  });
  test("ok, failed, failed, ok leaves exactly one line, for a writer and not a reader", () => {
    ok();
    fail("<b>status 500</b> " + "x".repeat(200));
    fail("status 502 is not 2xx");
    ok();
    const mine = lines("alice");
    expect(mine).toHaveLength(1);
    expect(mine[0]!.said!.startsWith("alice/hello sync failed: <b>status 500</b> ")).toBe(true);
    expect(mine[0]!.said!.length).toBeLessThanOrEqual("alice/hello sync failed: ".length + 120);
    expect(mine[0]!.link).toBe("/alice/hello/app");
    expect(lines("bob")).toHaveLength(0);
  });
  test("a failure after a recovery speaks again; the first ever pull failing speaks", () => {
    fail("status 503 is not 2xx");
    expect(lines("alice")).toHaveLength(2);
    expect(classifiedAudience("adapter.sync_failed")).toBe("writers");
  });
  // The terminal feed (`tui/screens/feed.ts`) calls `readFeed`, which reads `candidatesFor`.
  test("the terminal's feed read shows the line to a writer and not to a reader", () => {
    const seen = (name: string) =>
      readFeed(app.db, getUserByName(app.db, name)!, { limit: 50 }).items.filter((i) => i.what.includes("sync failed"));
    expect(seen("alice").length).toBeGreaterThan(0);
    expect(seen("bob")).toHaveLength(0);
  });
  test("the event is appended only once the run is committed", () => {
    let inTransaction: boolean | undefined;
    const bus = { append: (input: Parameters<App["bus"]["append"]>[0]) => ((inTransaction = app.db.inTransaction), app.bus.append(input)) };
    ok();
    recordPull(app.db, appId, { ok: false, reason: "late", durationMs: 1 }, bus);
    expect(inTransaction).toBe(false);
  });
});
		@@ -0,0 +1,18 @@
	1	# A failed adapter pull speaks once in the writers' feed (#10135)
	2
	3	When an app's adapter pull turns from ok (or from nothing) to failed, the
	4	forge appends one `adapter.sync_failed` event: "<owner>/<repo> sync failed:
	5	<reason>", linking to the app page. Repeated failures and recoveries append
	6	nothing, so a broken source says it once, not on every pull.
	7
	8	- Who sees it: the `writers` audience (`src/core/event_audience.ts`): those
	9	who may write the repository (`canWrite`). Readers and signed-out visitors
	10	see nothing.
	11	- Where: the home feed on the web, and the terminal's feed screen
	12	(`src/tui/screens/feed.ts`). Both call `readFeed`, which reads
	13	`candidatesFor` and its audience filter, so the two doors agree
	14	(`test/feed-sync-failed.test.ts` checks both).
	15	- Reason: stored cut to 200 characters (`PULL_REASON_MAX`), shown cut to
	16	120 in the line.
	17	- Timing: the event is appended after the run's transaction commits, so a
	18	live viewer never hears of a pull that was rolled back.
		@@ -38,6 +38,6 @@ export async function pullNow(
38	38	if (!writer.ok) return writer;
39	39	const record = getAppByRepo(app.db, repo.id);
40	40	if (!record) return err("not_found");
41		const pulled = await pullAppNow(app.db, record.id, reachOf(record));
	41	const pulled = await pullAppNow(app.db, record.id, reachOf(record), Date.now, app.bus);
42	42	return pulled.ok ? ok(pulled.run) : err(pulled.why);
43	43	}
		@@ -9,7 +9,9 @@
9	9	*/
10	10
11	11	import type { Database } from "bun:sqlite";
12		import { createPullRun, lastPullCursor, listPullRuns, type PullRun } from "../core/runs.ts";
	12	import { getAppById } from "../core/apps.ts";
	13	import { EventBus } from "../core/events.ts";
	14	import { createPullRun, lastPullCursor, lastPullState, listPullRuns, type PullRun } from "../core/runs.ts";
13	15
14	16	/** A body over this many bytes is a failed pull, never parsed. */
15	17	export const PULL_BODY_MAX = 64 * 1024;
		@@ -18,7 +20,7 @@ export const PULL_COUNT_MAX = 1_000_000;
18	20	/** The cursor is at most this many characters (code points). */
19	21	export const PULL_CURSOR_MAX = 200;
20	22	/** A failed pull's reason is cut to this many characters. */
21		const REASON_MAX = 200;
	23	export const PULL_REASON_MAX = 200;
22	24
23	25	const FIELDS = ["pulled", "skipped", "cursor", "errors"] as const;
24	26	const COUNTS = ["pulled", "skipped", "errors"] as const;
		@@ -39,7 +41,7 @@ export type PullAnswer =
39	41	/** An answer and how long the request took, as `recordPull` takes it. */
40	42	export type PullOutcome = PullAnswer & { readonly durationMs: number };
41	43
42		const fail = (reason: string): PullAnswer => ({ ok: false, reason: reason.slice(0, REASON_MAX) });
	44	const fail = (reason: string): PullAnswer => ({ ok: false, reason: reason.slice(0, PULL_REASON_MAX) });
43	45
44	46	/**
45	47	* Any control, format, private-use, unassigned or surrogate character, or a
		@@ -110,13 +112,26 @@ const durationOf = (ms: number): number => (Number.isFinite(ms) && ms > 0 ? Math
110	112	* the last good cursor stays where the next pull starts. A good answer whose
111	113	* cursor equals the last good pull's is a repeat: it is recorded, with its
112	114	* `pulled` counted as skipped and nothing written, so the same records never
113		* count as written twice. The read and the write are one transaction, so two
114		* recordings of the same answer cannot both see a fresh cursor.
	115	* count as written twice. A failed pull after an ok one (or the first) also
	116	* appends one `adapter.sync_failed` event, after the transaction commits, so
	117	* a live viewer never hears of a run that rolled back; repeats and recoveries
	118	* append none. The read and the write are one transaction, so two recordings
	119	* of the same answer cannot both see a fresh cursor.
115	120	*/
		junipercrew 2d ago outdated Re-wrap this comment; the line is far over the width the others keep.
116		export function recordPull(db: Database, appId: string, outcome: PullOutcome): PullRun {
	121	export function recordPull(
	122	db: Database,
	123	appId: string,
	124	outcome: PullOutcome,
	125	bus: Pick<EventBus, "append"> = new EventBus(db),
	126	): PullRun {
117	127	const durationMs = durationOf(outcome.durationMs);
118		return db.transaction((): PullRun => {
	128	let speak: { repoId: string; reason: string } \| undefined;
	129	const run = db.transaction((): PullRun => {
119	130	if (!outcome.ok) {
	131	// Only the turn from ok (or from nothing) to failed speaks.
	132	const record = lastPullState(db, appId) !== "failed" ? getAppById(db, appId) : undefined;
	133	const reason = outcome.reason.slice(0, PULL_REASON_MAX);
	134	if (record !== undefined) speak = { repoId: record.repoId, reason };
120	135	return createPullRun(db, {
121	136	appId,
122	137	state: "failed",
		@@ -125,7 +140,7 @@ export function recordPull(db: Database, appId: string, outcome: PullOutcome): P
125	140	failed: 0,
126	141	cursor: "",
127	142	durationMs,
128		reason: outcome.reason.slice(0, REASON_MAX),
	143	reason,
129	144	repeat: false,
130	145	});
131	146	}
		@@ -142,6 +157,17 @@ export function recordPull(db: Database, appId: string, outcome: PullOutcome): P
142	157	repeat,
143	158	});
144	159	}).immediate();
	160	if (speak !== undefined) {
	161	bus.append({
	162	// The literal stays a literal: the audience-coverage scan
	163	// (`core/event_audience.ts`) reads event kinds from `append` calls, and
	164	// a constant would hide this one from it.
	165	kind: "adapter.sync_failed",
	166	repoId: speak.repoId,
	167	payload: { appId, reason: speak.reason },
	168	});
	169	}
	170	return run;
145	171	}
146	172
147	173	/** How many runs `pullRunLines` answers when asked for none in particular, and at most. */
		@@ -16,6 +16,7 @@
16	16	import type { Database } from "bun:sqlite";
17	17	import type { App } from "../app.ts";
18	18	import { listApps, type AppRecord } from "../core/apps.ts";
	19	import type { EventBus } from "../core/events.ts";
19	20	import { getRepoById } from "../core/repos.ts";
20	21	import { lastPullAt, type PullRun } from "../core/runs.ts";
21	22	import { resolveOid } from "../git/merge.ts";
		@@ -76,6 +77,7 @@ async function pullOnce<Held extends string = never>(
76	77	reach: PullReach,
77	78	now: () => number,
78	79	hold: (adapter: AdapterSpec) => Held \| undefined = () => undefined,
	80	bus?: Pick<EventBus, "append">,
79	81	): Promise<PullOnce<Held>> {
80	82	const port = reach.port();
81	83	if (port === undefined) return { ok: false, why: "asleep" };
		@@ -88,22 +90,34 @@ async function pullOnce<Held extends string = never>(
88	90	if (held !== undefined) return { ok: false, why: held };
89	91	const started = now();
90	92	const answer = answerOf(await reach.post(port));
91		return { ok: true, run: recordPull(db, appId, { ...answer, durationMs: Math.max(0, now() - started) }) };
	93	return { ok: true, run: recordPull(db, appId, { ...answer, durationMs: Math.max(0, now() - started) }, bus) };
92	94	} finally {
93	95	inFlight.delete(appId);
94	96	}
95	97	}
96	98
97	99	/** Pull one app if it is a running adapter and due: whether a pull was made. */
98		export async function pullApp(db: Database, appId: string, reach: PullReach, now: () => number = Date.now): Promise<boolean> {
	100	export async function pullApp(
	101	db: Database,
	102	appId: string,
	103	reach: PullReach,
	104	now: () => number = Date.now,
	105	bus?: Pick<EventBus, "append">,
	106	): Promise<boolean> {
99	107	const notDue = (adapter: AdapterSpec) =>
100	108	isPullDue(lastPullAt(db, appId), adapter.schedule.every, now()) ? undefined : ("not_due" as const);
101		return (await pullOnce(db, appId, reach, now, notDue)).ok;
	109	return (await pullOnce(db, appId, reach, now, notDue, bus)).ok;
102	110	}
103	111
104	112	/** Pull one app now, whatever its schedule (#10084): the recorded run, or why not. Never wakes it. */
105		export function pullAppNow(db: Database, appId: string, reach: PullReach, now: () => number = Date.now): Promise<PullOnce> {
106		return pullOnce(db, appId, reach, now);
	113	export function pullAppNow(
	114	db: Database,
	115	appId: string,
	116	reach: PullReach,
	117	now: () => number = Date.now,
	118	bus?: Pick<EventBus, "append">,
	119	): Promise<PullOnce> {
	120	return pullOnce(db, appId, reach, now, undefined, bus);
107	121	}
108	122
109	123	/** Who the forge says it is when it pulls: an admin of the app, named for itself. */
		@@ -203,7 +217,7 @@ export async function pullDue(
203	217	const pulled: string[] = [];
204	218	for (const record of listApps(app.db)) {
205	219	try {
206		if (await pullApp(app.db, record.id, reachOf(record), now)) pulled.push(record.slug);
	220	if (await pullApp(app.db, record.id, reachOf(record), now, app.bus)) pulled.push(record.slug);
207	221	} catch (error) {
208	222	app.log.warn("could not pull an adapter", { app: record.slug, error: String(error) });
209	223	}
		@@ -27,10 +27,10 @@ import type { Database } from "bun:sqlite";
27	27	import type { ForgeEvent } from "./events.ts";
28	28	import { getAppByRepo } from "./apps.ts";
29	29	import { personOwners } from "./orgs.ts";
30		import { canAdmin, canEnterApp, canSeeOwnerFacts, getRepoById } from "./repos.ts";
	30	import { canAdmin, canEnterApp, canWrite, canSeeOwnerFacts, getRepoById } from "./repos.ts";
31	31	import type { User } from "./users.ts";
32	32
33		export type EventAudience = "readers" \| "admins" \| "actor" \| "app";
	33	export type EventAudience = "readers" \| "admins" \| "actor" \| "app" \| "writers";
34	34
35	35	/**
36	36	* Whether this viewer answers for a deleted repository: the forge's admin, who
		@@ -243,6 +243,8 @@ export function classifiedAudience(kind: string): EventAudience \| undefined {
243	243	kind === "repo.file_released"
244	244	)
245	245	return "admins";
	246	// A broken adapter sync is for those who may write the repository (`apps/pull.ts`).
	247	if (kind === "adapter.sync_failed") return "writers";
246	248	// A fleet operation spans repositories; its receipt is its actor's. So is
247	249	// a thought: it belongs to no repository and to nobody but its author.
248	250	if (
		@@ -291,6 +293,15 @@ export function audienceFilter(
291	293	admins.set(repoId, yes);
292	294	return yes;
293	295	};
	296	const writers = new Map<string, boolean>();
	297	const mayWrite = (repoId: string): boolean => {
	298	const known = writers.get(repoId);
	299	if (known !== undefined) return known;
	300	const repo = getRepoById(db, repoId);
	301	const yes = repo !== undefined && canWrite(db, repo, viewer);
	302	writers.set(repoId, yes);
	303	return yes;
	304	};
294	305	const entrants = new Map<string, boolean>();
295	306	const mayEnter = (repoId: string): boolean => {
296	307	const known = entrants.get(repoId);
		@@ -348,6 +359,8 @@ export function audienceFilter(
348	359	return true;
349	360	case "app":
350	361	return event.repoId === null \|\| mayEnter(event.repoId);
	362	case "writers":
	363	return event.repoId !== null && mayWrite(event.repoId);
351	364	case "actor":
352	365	return viewer !== undefined && event.actorId === viewer.id;
353	366	case "admins":
		@@ -1588,6 +1588,17 @@ export function listPullRuns(db: Database, appId: string, limit = 50): PullRun[]
1588	1588	.map(toPullRun);
1589	1589	}
1590	1590
	1591	/** The state of an app's newest pull run; undefined before the first. */
	1592	export function lastPullState(db: Database, appId: string): "ok" \| "failed" \| undefined {
	1593	return (
	1594	db
	1595	.query<{ state: "ok" \| "failed" }, [string]>(
	1596	`SELECT state FROM pull_run WHERE app_id = ?1 ORDER BY id DESC LIMIT 1`,
	1597	)
	1598	.get(appId)?.state ?? undefined
	1599	);
	1600	}
	1601
1591	1602	/** When an app's last pull, good or failed, was recorded, in ms since the epoch; undefined before the first. */
1592	1603	export function lastPullAt(db: Database, appId: string): number \| undefined {
1593	1604	const row = db
		@@ -161,6 +161,8 @@ export function quietRunsFor(
161	161	.filter((e) => quiet.has(e.id) && repos.get(e.repoId!)?.archivedAt == null)
162	162	.map((e) => ({ repoId: e.repoId!, actorId: e.actorId, id: e.id }));
163	163	}
	164	/** How much of a sync failure's stored reason (`PULL_REASON_MAX`, `apps/pull.ts`) a feed line shows. */
	165	const SYNC_REASON_SHOWN = 120;
164	166	const clip = (s: string, n = 120): string =>
165	167	s.length > n ? `${s.slice(0, n - 1)}…` : s;
166	168
		@@ -630,6 +632,17 @@ function fromEvent(
630	632	what: `${/^[aeiou]/.test(kind) ? "an" : "a"} ${kind} run${subject} ${state === "failed" ? "failed" : "finished"}${state === "failed" ? (failure ? `: ${failure}` : "") : summary ? `: ${summary}` : ""}`,
631	633	};
632	634	}
	635	case "adapter.sync_failed": {
	636	const reason = clip(typeof p["reason"] === "string" ? p["reason"] : "", SYNC_REASON_SHOWN);
	637	return {
	638	...common,
	639	kind: "run",
	640	link: `${base(repo)}/app`,
	641	urgency: 4,
	642	said: `${slug(repo)} sync failed: ${reason}`,
	643	what: `sync failed: ${reason}`,
	644	};
	645	}
633	646	case "pull.opened":
634	647	return { ...common, kind: "pull", link: change, what: `opened a pull request${title ? `: ${clip(title)}` : ""}${toldWords}` };
635	648	case "pull.merged":
		@@ -0,0 +1,81 @@
	1	/** A pull turning failed speaks once in its writers' feed, and only to them (#10135). */
	2	import { afterAll, beforeAll, describe, expect, test } from "bun:test";
	3	import { mkdtemp, rm } from "node:fs/promises";
	4	import { tmpdir } from "node:os";
	5	import { join } from "node:path";
	6	import { createApp, type App } from "../src/app.ts";
	7	import { newApp } from "../src/actions/apps.ts";
	8	import { recordPull } from "../src/apps/pull.ts";
	9	import { classifiedAudience } from "../src/core/event_audience.ts";
	10	import { getRepo } from "../src/core/repos.ts";
	11	import { createUser, getUserByName } from "../src/core/users.ts";
	12	import { readFeed } from "../src/feed/feed.ts";
	13	import { candidatesFor } from "../src/feed/items.ts";
	14	import { seedAdmin } from "./helpers/admin.ts";
	15
	16	describe("sync failed feed line", () => {
	17	let dataDir: string;
	18	let app: App;
	19	let appId: string;
	20
	21	const ok = () => recordPull(app.db, appId, { ok: true, pulled: 1, skipped: 0, errors: 0, cursor: `c${Math.random()}`, durationMs: 1 }, app.bus);
	22	const fail = (reason: string) => recordPull(app.db, appId, { ok: false, reason, durationMs: 1 }, app.bus);
	23	const lines = (name: string) => {
	24	const viewer = getUserByName(app.db, name)!;
	25	return candidatesFor(app.db, viewer, Date.now(), undefined, false).filter((c) => c.said?.includes("sync failed"));
	26	};
	27
	28	beforeAll(async () => {
	29	dataDir = await mkdtemp(join(tmpdir(), "forge-sync-failed-"));
	30	app = createApp({ dataDir, port: 0 });
	31	await seedAdmin(app, { username: "alice", email: "[email protected]", password: "a-good-passphrase" });
	32	const b = await createUser(app.db, {
	33	origin: "seed", trusted: true, username: "bob",
	34	email: "[email protected]", password: "a-good-passphrase",
	35	});
	36	if (!b.ok) throw new Error(b.error);
	37	const made = await newApp(app, getUserByName(app.db, "alice")!, { name: "hello", visibility: "public" } as never);
	38	if (!made.ok) throw new Error(String(made.error));
	39	const repo = getRepo(app.db, "alice", "hello")!;
	40	appId = (app.db.query("SELECT id FROM app WHERE repo_id = ?1").get(repo.id) as { id: string }).id;
	41	});
	42	afterAll(async () => {
	43	app.db.close();
	44	await rm(dataDir, { recursive: true, force: true });
	45	});
	46
	47	test("ok, failed, failed, ok leaves exactly one line, for a writer and not a reader", () => {
	48	ok();
	49	fail("<b>status 500</b> " + "x".repeat(200));
	50	fail("status 502 is not 2xx");
	51	ok();
	52	const mine = lines("alice");
	53	expect(mine).toHaveLength(1);
	54	expect(mine[0]!.said!.startsWith("alice/hello sync failed: <b>status 500</b> ")).toBe(true);
	55	expect(mine[0]!.said!.length).toBeLessThanOrEqual("alice/hello sync failed: ".length + 120);
	56	expect(mine[0]!.link).toBe("/alice/hello/app");
	57	expect(lines("bob")).toHaveLength(0);
	58	});
	59
	60	test("a failure after a recovery speaks again; the first ever pull failing speaks", () => {
	61	fail("status 503 is not 2xx");
	62	expect(lines("alice")).toHaveLength(2);
	63	expect(classifiedAudience("adapter.sync_failed")).toBe("writers");
	64	});
	65
	66	// The terminal feed (`tui/screens/feed.ts`) calls `readFeed`, which reads `candidatesFor`.
	67	test("the terminal's feed read shows the line to a writer and not to a reader", () => {
	68	const seen = (name: string) =>
	69	readFeed(app.db, getUserByName(app.db, name)!, { limit: 50 }).items.filter((i) => i.what.includes("sync failed"));
	70	expect(seen("alice").length).toBeGreaterThan(0);
	71	expect(seen("bob")).toHaveLength(0);
	72	});
	73
	74	test("the event is appended only once the run is committed", () => {
	75	let inTransaction: boolean \| undefined;
	76	const bus = { append: (input: Parameters<App["bus"]["append"]>[0]) => ((inTransaction = app.db.inTransaction), app.bus.append(input)) };
	77	ok();
	78	recordPull(app.db, appId, { ok: false, reason: "late", durationMs: 1 }, bus);
	79	expect(inTransaction).toBe(false);
	80	});
	81	});