Notify on failures
diff --git a/client/src/styles/app.css b/client/src/styles/app.css
index f28084c..cd589b2 100644
--- a/client/src/styles/app.css
+++ b/client/src/styles/app.css
@@ -1682,3 +1682,96 @@ a {
color: var(--color-text-muted);
font-size: 0.8rem;
}
+
+/* ---------- Live update progress (header) ---------- */
+.header-progress {
+ display: inline-flex;
+ align-items: center;
+ gap: 6px;
+ margin-left: 8px;
+ padding: 3px 10px;
+ border-radius: 999px;
+ background: var(--color-accent);
+ color: var(--color-accent-contrast);
+ font-size: 0.75rem;
+ font-weight: 600;
+ white-space: nowrap;
+ text-decoration: none;
+}
+
+.header-progress .spinner {
+ width: 10px;
+ height: 10px;
+ border-width: 2px;
+ border-color: currentColor;
+ border-right-color: transparent;
+}
+
+/* ---------- Card: download size, footer options, container logs ---------- */
+.download-size {
+ font-size: 0.75rem;
+ color: var(--color-text-muted);
+ white-space: nowrap;
+}
+
+.card-footer-row {
+ display: flex;
+ align-items: flex-start;
+ justify-content: space-between;
+ gap: 8px;
+ flex-wrap: wrap;
+}
+
+.card-footer-row > div:first-child {
+ flex: 1 1 100%;
+ order: 2;
+}
+
+.card-footer-actions {
+ display: flex;
+ align-items: center;
+ gap: 12px;
+ flex-wrap: wrap;
+ order: 1;
+ width: 100%;
+}
+
+.card-option {
+ display: inline-flex;
+ align-items: center;
+ gap: 6px;
+ font-size: 0.8rem;
+ color: var(--color-text-muted);
+ cursor: pointer;
+}
+
+.card-option input {
+ accent-color: var(--color-accent);
+}
+
+.container-logs-head {
+ display: flex;
+ align-items: center;
+ justify-content: space-between;
+ gap: 8px;
+ font-size: 0.8rem;
+ color: var(--color-text-muted);
+ margin-bottom: 6px;
+}
+
+.container-logs-body {
+ max-height: 320px;
+ overflow: auto;
+ font-size: 0.72rem;
+}
+
+.history-row-log {
+ margin-top: 8px;
+}
+
+.history-log-body {
+ max-height: 320px;
+ overflow: auto;
+ font-size: 0.72rem;
+ cursor: text;
+}
diff --git a/server/src/checker.js b/server/src/checker.js
index c897fbd..1302226 100644
--- a/server/src/checker.js
+++ b/server/src/checker.js
@@ -8,7 +8,7 @@
*/
import { listContainers } from './docker.js';
-import { getRemoteDigest, getRemoteVersion, listTags } from './registry.js';
+import { getRemoteDigest, getRemoteVersion, listTags, getDownloadSize } from './registry.js';
import { digestsEqual, isRunningDigest } from './reconcile.js';
import { isMeaningfulVersion, parseVersionTag, findNewerTags } from './version.js';
import { parseRef } from './reconcile.js';
@@ -202,6 +202,10 @@ async function doCheck() {
// repeated checks).
const existing = db.latestUnresolvedEventForRef(c.normalizedRef);
if (existing && digestsEqual(existing.digest, remote)) {
+ // Backfill the download size for updates found before it was tracked.
+ if (existing.download_bytes == null) {
+ db.setEventDownloadSize(c.normalizedRef, remote, await getDownloadSize(c.image, remote, c.currentDigest));
+ }
// Already flagged. If we previously stored a junk version label
// (e.g. "main"), try to backfill a real one now without waiting for
// a new image to appear.
@@ -234,6 +238,8 @@ async function doCheck() {
breaking,
raw_json: JSON.stringify({ source: 'check' }),
});
+ // How much the update will download (new layers only). Best-effort.
+ db.setEventDownloadSize(c.normalizedRef, remote, await getDownloadSize(c.image, remote, c.currentDigest));
// Remember versions per digest: the available one keyed by the remote
// digest (so it shows instantly once the user updates), and the running
// one if its own label is usable.
diff --git a/server/src/containers-service.js b/server/src/containers-service.js
index 347afb0..ae27c27 100644
--- a/server/src/containers-service.js
+++ b/server/src/containers-service.js
@@ -39,6 +39,7 @@ export function buildContainerItems({
getRollback = () => null,
getCheckError = () => null,
getTagUpdate = () => null,
+ getAutoRevert = () => false,
tagPolicy = 'minor',
}) {
const items = [];
@@ -107,10 +108,15 @@ export function buildContainerItems({
// must not leak through once the digests match again.
breakingRisk: !!(updateAvailable && event?.breaking),
skipped,
+ // Bytes the update will download (new layers only), when known.
+ downloadSize: (updateAvailable || skipped) && event?.download_bytes != null
+ ? { bytes: event.download_bytes, exact: Boolean(event.download_exact) }
+ : null,
newerTag,
newerMajorTag,
pinned: isPinned(c.normalizedRef),
canRevert: !!rollback,
+ autoRevert: getAutoRevert(c.name),
rollbackVersion: rollback?.old_version ?? null,
checkError: getCheckError(c.normalizedRef),
state: c.state,
diff --git a/server/src/db.js b/server/src/db.js
index ed17c6a..a4df22d 100644
--- a/server/src/db.js
+++ b/server/src/db.js
@@ -82,6 +82,10 @@ CREATE INDEX IF NOT EXISTS idx_history_created ON update_history(created_at DESC
if (!cols.includes('available_version')) {
db.exec('ALTER TABLE update_events ADD COLUMN available_version TEXT');
}
+ if (!cols.includes('download_bytes')) {
+ db.exec('ALTER TABLE update_events ADD COLUMN download_bytes INTEGER');
+ db.exec('ALTER TABLE update_events ADD COLUMN download_exact INTEGER');
+ }
if (!cols.includes('breaking')) {
db.exec('ALTER TABLE update_events ADD COLUMN breaking INTEGER DEFAULT 0');
}
@@ -99,6 +103,17 @@ CREATE INDEX IF NOT EXISTS idx_history_created ON update_history(created_at DESC
if (!cols.includes('compose_edit')) db.exec('ALTER TABLE rollback_points ADD COLUMN compose_edit TEXT');
}
+// update_history.log: the tail of the update's live log, so a failure can be
+// investigated after the fact. container_prefs: per-container options.
+{
+ const cols = db.prepare('PRAGMA table_info(update_history)').all().map((col) => col.name);
+ if (!cols.includes('log')) db.exec('ALTER TABLE update_history ADD COLUMN log TEXT');
+ db.exec(`CREATE TABLE IF NOT EXISTS container_prefs (
+ container_name TEXT PRIMARY KEY,
+ auto_revert INTEGER DEFAULT 0
+ )`);
+}
+
// update_history: remember versions on the row itself, for when an image's
// digest is unknown (so the digest→version lookup can't recover them later).
{
@@ -108,6 +123,12 @@ CREATE INDEX IF NOT EXISTS idx_history_created ON update_history(created_at DESC
}
const stmts = {
+ getHistoryLog: db.prepare(`SELECT log FROM update_history WHERE id = ? LIMIT 1`),
+ getAutoRevert: db.prepare(`SELECT auto_revert FROM container_prefs WHERE container_name = ? LIMIT 1`),
+ setAutoRevert: db.prepare(`
+ INSERT INTO container_prefs (container_name, auto_revert) VALUES (?, ?)
+ ON CONFLICT(container_name) DO UPDATE SET auto_revert = excluded.auto_revert
+ `),
importHistoryRow: db.prepare(`
INSERT INTO update_history (container_name, image, old_digest, new_digest, old_version, new_version, status, message, created_at)
VALUES (@container_name, @image, @old_digest, @new_digest, @old_version, @new_version, @status, @message, @created_at)
@@ -148,6 +169,10 @@ const stmts = {
ORDER BY id DESC LIMIT 1
)
`),
+ setEventDownloadSize: db.prepare(`
+ UPDATE update_events SET download_bytes = ?, download_exact = ?
+ WHERE normalized_ref = ? AND digest = ? AND resolved = 0
+ `),
updateEventAvailableVersion: db.prepare(`
UPDATE update_events SET available_version = ?
WHERE normalized_ref = ? AND digest = ? AND resolved = 0
@@ -188,16 +213,16 @@ const stmts = {
DELETE FROM rollback_points WHERE container_name = ?
`),
recordUpdate: db.prepare(`
- INSERT INTO update_history (container_name, image, old_digest, new_digest, old_version, new_version, status, message)
- VALUES (@container_name, @image, @old_digest, @new_digest, @old_version, @new_version, @status, @message)
+ INSERT INTO update_history (container_name, image, old_digest, new_digest, old_version, new_version, status, message, log)
+ VALUES (@container_name, @image, @old_digest, @new_digest, @old_version, @new_version, @status, @message, @log)
`),
getHistoryAll: db.prepare(`
- SELECT * FROM update_history
+ SELECT id, container_name, image, old_digest, new_digest, old_version, new_version, status, message, created_at, (log IS NOT NULL AND log != '') AS has_log FROM update_history
ORDER BY created_at DESC, id DESC
LIMIT ? OFFSET ?
`),
getHistoryByContainer: db.prepare(`
- SELECT * FROM update_history
+ SELECT id, container_name, image, old_digest, new_digest, old_version, new_version, status, message, created_at, (log IS NOT NULL AND log != '') AS has_log FROM update_history
WHERE container_name = ?
ORDER BY created_at DESC, id DESC
LIMIT ? OFFSET ?
@@ -290,6 +315,11 @@ export function markTagNotified(normalized_ref, key) {
return stmts.markTagNotified.run(key ?? null, normalized_ref);
}
+export function setEventDownloadSize(normalized_ref, digest, size) {
+ if (!size) return undefined;
+ return stmts.setEventDownloadSize.run(size.bytes, size.exact ? 1 : 0, normalized_ref, digest);
+}
+
export function updateEventAvailableVersion(normalized_ref, digest, available_version) {
return stmts.updateEventAvailableVersion.run(available_version ?? null, normalized_ref, digest);
}
@@ -384,7 +414,7 @@ export function getAllRollbackPoints() {
return stmts.getAllRollbackPoints.all();
}
-export function recordUpdate({ container_name, image, old_digest, new_digest, old_version, new_version, status, message }) {
+export function recordUpdate({ container_name, image, old_digest, new_digest, old_version, new_version, status, message, log }) {
return stmts.recordUpdate.run({
container_name,
image,
@@ -394,9 +424,24 @@ export function recordUpdate({ container_name, image, old_digest, new_digest, ol
new_version: new_version ?? null,
status,
message: message ?? null,
+ log: log ? String(log).slice(-50_000) : null,
});
}
+/** The saved log for one history entry, or null. */
+export function getHistoryLog(id) {
+ return stmts.getHistoryLog.get(id)?.log ?? null;
+}
+
+/** Per-container "revert automatically if the update comes up unhealthy". */
+export function getAutoRevert(container_name) {
+ return Boolean(stmts.getAutoRevert.get(container_name)?.auto_revert);
+}
+
+export function setAutoRevert(container_name, enabled) {
+ return stmts.setAutoRevert.run(container_name, enabled ? 1 : 0);
+}
+
export function getHistory({ containerName, limit = 50, offset = 0 } = {}) {
if (containerName) {
return stmts.getHistoryByContainer.all(containerName, limit, offset);
diff --git a/server/src/docker.js b/server/src/docker.js
index 2da3394..964a3a8 100644
--- a/server/src/docker.js
+++ b/server/src/docker.js
@@ -152,10 +152,19 @@ export function pickRepoDigests(repoDigests, image) {
* @param {string} image - configured image ref, e.g. "nginx:latest".
* @returns {Promise<{ digest: string|null, digests: string[], version: string|null, source: string|null }>}
*/
-async function inspectImageMeta(imageIdOrName, image) {
+async function inspectImageMeta(imageIdOrName, image, cache = null) {
let imageInfo;
try {
- imageInfo = await docker.getImage(imageIdOrName).inspect();
+ if (cache) {
+ if (!cache.has(imageIdOrName)) {
+ const p = docker.getImage(imageIdOrName).inspect();
+ p.catch(() => {}); // a failure is handled by whichever caller awaits it
+ cache.set(imageIdOrName, p);
+ }
+ imageInfo = await cache.get(imageIdOrName);
+ } else {
+ imageInfo = await docker.getImage(imageIdOrName).inspect();
+ }
} catch (err) {
console.warn(`docker.js: failed to inspect image ${imageIdOrName}: ${err.message}`);
return { digest: null, digests: [], version: null, source: null };
@@ -323,6 +332,24 @@ async function tagPointsElsewhere(ref, imageId, cache) {
return Boolean(tagId) && tagId !== imageId;
}
+/**
+ * Pure helper: map `items` through async `fn` with at most `limit` in flight,
+ * preserving order.
+ */
+export async function mapLimit(items, limit, fn) {
+ const out = new Array(items.length);
+ let next = 0;
+ const worker = async () => {
+ while (next < items.length) {
+ const i = next;
+ next += 1;
+ out[i] = await fn(items[i], i);
+ }
+ };
+ await Promise.all(Array.from({ length: Math.min(limit, items.length) }, worker));
+ return out;
+}
+
/**
* Lists all containers (including stopped ones) with the fields needed by
* the `/api/containers` endpoint's docker-derived data. Skips (with a
@@ -337,10 +364,10 @@ async function tagPointsElsewhere(ref, imageId, cache) {
*/
export async function listContainers() {
const summaries = await docker.listContainers({ all: true });
- const results = [];
const tagIdCache = new Map(); // image ref -> local image ID it points at
+ const imageCache = new Map(); // image ID -> image inspect (shared by containers)
- for (const summary of summaries) {
+ const inspectOne = async (summary) => {
try {
const container = docker.getContainer(summary.Id);
const inspectData = await container.inspect();
@@ -350,16 +377,16 @@ export async function listContainers() {
// Never list our own container — offering to update it would recreate
// the container running this process mid-update.
if (isSelfContainer(name, summary.Id)) {
- continue;
+ return null;
}
const image = trackedImageRef(inspectData);
if (!image) {
console.warn(`docker.js: container ${name} has no Config.Image, skipping`);
- continue;
+ return null;
}
- const meta = await inspectImageMeta(inspectData.Image, image);
+ const meta = await inspectImageMeta(inspectData.Image, image, imageCache);
const { version: currentVersion, source: sourceUrl } = meta;
let currentDigests = meta.digests;
// A reverted container runs an untagged image with no RepoDigests; use
@@ -402,7 +429,7 @@ export async function listContainers() {
tag = parseRef(image).tag;
} catch (err) {
console.warn(`docker.js: failed to normalize ref "${image}" for ${name}: ${err.message}`);
- continue;
+ return null;
}
// Flag compose-managed containers whose compose file isn't reachable
@@ -410,7 +437,7 @@ export async function listContainers() {
// the dashboard can warn before an update is attempted.
const composeFileMissing = Boolean(composeFile) && !fs.existsSync(composeFile);
- results.push({
+ return {
name,
image,
tag,
@@ -425,14 +452,16 @@ export async function listContainers() {
workingDir: workingDir || null,
state: inspectData.State?.Status || summary.State || 'unknown',
normalizedRef,
- });
+ };
} catch (err) {
console.warn(`docker.js: failed to inspect container ${summary.Id}: ${err.message}`);
- // skip this container, continue with the rest
+ return null; // skip this container, continue with the rest
}
- }
+ };
- return results;
+ // Inspect several containers at once (one-by-one made the dashboard slow on
+ // hosts with many containers), keeping Docker's order.
+ return (await mapLimit(summaries, 8, inspectOne)).filter(Boolean);
}
/**
@@ -1227,6 +1256,52 @@ export async function getContainerImageMeta(name) {
return { image, currentVersion: version, sourceUrl: source || knownSourceFor(image) };
}
+/**
+ * Pure: decode Docker's multiplexed log stream (used when a container has no
+ * TTY): frames of [stream byte, 0, 0, 0, uint32 BE length, payload]. A TTY
+ * container's logs are raw text instead; detect that by the header shape.
+ *
+ * @param {Buffer} buf
+ * @returns {string}
+ */
+export function demuxDockerLogs(buf) {
+ if (!Buffer.isBuffer(buf)) return String(buf ?? '');
+ const looksFramed =
+ buf.length >= 8 && buf[0] <= 2 && buf[1] === 0 && buf[2] === 0 && buf[3] === 0;
+ if (!looksFramed) return buf.toString('utf8');
+ const parts = [];
+ let i = 0;
+ while (i + 8 <= buf.length) {
+ const len = buf.readUInt32BE(i + 4);
+ parts.push(buf.subarray(i + 8, i + 8 + len));
+ i += 8 + len;
+ }
+ return Buffer.concat(parts).toString('utf8');
+}
+
+/**
+ * The last `tail` lines of a container's own output (stdout + stderr), e.g. to
+ * see why an update crash-loops. Capped to keep responses small.
+ *
+ * @returns {Promise<{ text: string, state: string, restartCount: number }>}
+ */
+export async function getContainerLogs(name, tail = 200) {
+ const container = docker.getContainer(name);
+ const info = await container.inspect();
+ const raw = await container.logs({
+ stdout: true,
+ stderr: true,
+ timestamps: true,
+ tail: Math.min(Math.max(1, tail), 1000),
+ });
+ const text = info.Config?.Tty ? Buffer.from(raw).toString('utf8') : demuxDockerLogs(Buffer.from(raw));
+ return {
+ text: text.slice(-200_000),
+ state: info.State?.Status || 'unknown',
+ restartCount: info.RestartCount ?? 0,
+ };
+}
+
/**
* Normalizes a Docker image ID (`sha256:<64-hex>` or already-short) down to
* the 12-char short form used for display and for matching a dangling image
diff --git a/server/src/jobs.js b/server/src/jobs.js
new file mode 100644
index 0000000..089c2b4
--- /dev/null
+++ b/server/src/jobs.js
@@ -0,0 +1,226 @@
+/**
+ * Server-side job queue for updates, reverts and tag switches.
+ *
+ * Updates used to be driven by the browser: "Update all" was a loop in the
+ * dashboard page, so navigating away, locking the phone or a card scrolling
+ * out of view stopped the batch part-way, and coming back showed nothing of
+ * what was still running. Now every job runs here, independent of any page:
+ *
+ * - Jobs in the same stack (compose project) run one at a time — concurrent
+ * `docker compose up` calls against one project race on its networks and
+ * dependencies — while different stacks run in parallel, at most
+ * MAX_PARALLEL at once (to not saturate the host's bandwidth/disk).
+ * - Each container has at most one queued/running job (a second is refused).
+ * - Finished jobs are remembered for RETAIN_MS so a dashboard that comes back
+ * can show what happened (and replay the log — see sse.js).
+ * - A batch ("Update all") is tracked as a whole, and `onBatchDone` fires once
+ * when its last job finishes (used for the summary notification).
+ *
+ * The work itself is injected (`execute`), so this module is pure scheduling
+ * and fully unit-testable. Jobs live in memory: a server restart forgets them.
+ */
+
+const MAX_PARALLEL = 3;
+const RETAIN_MS = 15 * 60_000;
+
+const jobs = new Map(); // container name -> latest job
+const lanes = new Map(); // lane key -> tail promise of that lane
+const batches = new Map(); // batch id -> batch
+let running = 0;
+const waiting = []; // resolvers waiting for a parallel slot
+let seq = 0;
+const listeners = { change: [], batchDone: [] };
+
+function emit(kind, payload) {
+ for (const fn of listeners[kind]) {
+ try {
+ fn(payload);
+ } catch (err) {
+ console.warn(`jobs.js: ${kind} listener threw: ${err.message}`);
+ }
+ }
+}
+
+/** Subscribe to queue changes (enqueue / start / finish). */
+export function onChange(fn) {
+ listeners.change.push(fn);
+}
+
+/** Subscribe to "a batch finished" (receives the batch summary). */
+export function onBatchDone(fn) {
+ listeners.batchDone.push(fn);
+}
+
+async function acquireSlot() {
+ if (running < MAX_PARALLEL) {
+ running += 1;
+ return;
+ }
+ await new Promise((resolve) => waiting.push(resolve));
+ running += 1;
+}
+
+function releaseSlot() {
+ running -= 1;
+ const next = waiting.shift();
+ if (next) next();
+}
+
+/** True if `name` has a queued or running job. */
+export function isBusy(name) {
+ const j = jobs.get(name);
+ return Boolean(j && (j.status === 'queued' || j.status === 'running'));
+}
+
+function publicJob(j) {
+ const { execute, ...rest } = j;
+ return rest;
+}
+
+/**
+ * Queue a job.
+ *
+ * @param {object} spec
+ * @param {string} spec.name - container name.
+ * @param {string} spec.kind - 'update' | 'revert' | 'switch'.
+ * @param {string|null} [spec.project] - compose project (lane); null = own lane.
+ * @param {string|null} [spec.batchId]
+ * @param {object} [spec.meta] - extra fields shown to clients (e.g. tag).
+ * @param {(job) => Promise<{ success: boolean, message: string }>} spec.execute
+ * @returns {object} the job (public fields).
+ * @throws {Error} with `.code = 'busy'` if the container already has a job.
+ */
+export function enqueue({ name, kind, project = null, batchId = null, meta = {}, execute }) {
+ if (isBusy(name)) {
+ const err = new Error(`${name} already has an update queued or running`);
+ err.code = 'busy';
+ throw err;
+ }
+ const job = {
+ id: ++seq,
+ name,
+ kind,
+ project,
+ batchId,
+ ...meta,
+ status: 'queued',
+ queuedAt: Date.now(),
+ startedAt: null,
+ finishedAt: null,
+ success: null,
+ message: null,
+ execute,
+ };
+ jobs.set(name, job);
+ if (batchId && batches.has(batchId)) batches.get(batchId).names.push(name);
+
+ const laneKey = project ? `p:${project}` : `c:${name}`;
+ const prev = lanes.get(laneKey) || Promise.resolve();
+ // Why it's waiting, for the UI: another update in its stack, or just the
+ // parallel cap.
+ job.waitingOn = lanes.has(laneKey) ? 'stack' : running >= MAX_PARALLEL ? 'slot' : null;
+ const tail = prev.then(async () => {
+ await acquireSlot();
+ job.status = 'running';
+ job.waitingOn = null;
+ job.startedAt = Date.now();
+ emit('change', publicJob(job));
+ let outcome;
+ try {
+ outcome = await job.execute(job);
+ } catch (err) {
+ outcome = { success: false, message: err.message };
+ } finally {
+ releaseSlot();
+ }
+ job.status = 'done';
+ job.finishedAt = Date.now();
+ job.success = Boolean(outcome?.success);
+ job.message = outcome?.message ?? '';
+ emit('change', publicJob(job));
+ settleBatch(job.batchId);
+ });
+ lanes.set(laneKey, tail);
+ tail.finally(() => {
+ if (lanes.get(laneKey) === tail) lanes.delete(laneKey);
+ });
+
+ emit('change', publicJob(job));
+ return publicJob(job);
+}
+
+/** Start a batch; enqueue its jobs with the returned id. */
+export function startBatch() {
+ const id = `b${Date.now().toString(36)}${(++seq).toString(36)}`;
+ batches.set(id, { id, names: [], startedAt: Date.now(), finishedAt: null });
+ return id;
+}
+
+function batchSummary(b) {
+ const list = b.names.map((n) => jobs.get(n)).filter((j) => j && j.batchId === b.id);
+ const done = list.filter((j) => j.status === 'done');
+ return {
+ id: b.id,
+ total: list.length,
+ done: done.length,
+ succeeded: done.filter((j) => j.success).map((j) => j.name),
+ failed: done.filter((j) => !j.success).map((j) => ({ name: j.name, message: j.message })),
+ startedAt: b.startedAt,
+ finishedAt: b.finishedAt,
+ };
+}
+
+/** Close a batch with no more jobs to add (fires onBatchDone if all are already done). */
+export function sealBatch(batchId) {
+ const b = batches.get(batchId);
+ if (!b) return;
+ b.sealed = true;
+ settleBatch(batchId);
+}
+
+function settleBatch(batchId) {
+ const b = batchId && batches.get(batchId);
+ if (!b || !b.sealed || b.finishedAt) return;
+ const s = batchSummary(b);
+ if (s.done < s.total) return;
+ b.finishedAt = Date.now();
+ emit('batchDone', batchSummary(b));
+ emit('change', null);
+}
+
+/**
+ * Current state for clients: queued/running jobs plus those finished within
+ * the retention window, and the most recent batch (if still relevant).
+ */
+export function snapshot(now = Date.now()) {
+ const list = [];
+ for (const [name, j] of jobs) {
+ if (j.status === 'done' && now - j.finishedAt > RETAIN_MS) {
+ jobs.delete(name);
+ continue;
+ }
+ list.push(publicJob(j));
+ }
+ let latest = null;
+ for (const [id, b] of batches) {
+ if (b.finishedAt && now - b.finishedAt > RETAIN_MS) {
+ batches.delete(id);
+ continue;
+ }
+ if (!latest || b.startedAt > latest.startedAt) latest = b;
+ }
+ return { jobs: list.sort((a, b) => a.queuedAt - b.queuedAt), batch: latest ? batchSummary(latest) : null };
+}
+
+/** Test hook: wait until every lane is idle. */
+export async function _drain() {
+ while (lanes.size) await Promise.all([...lanes.values()]);
+}
+
+/** Test hook: forget everything. */
+export function _reset() {
+ jobs.clear();
+ batches.clear();
+}
+
+export default { enqueue, isBusy, startBatch, sealBatch, snapshot, onChange, onBatchDone };
diff --git a/server/src/notify.js b/server/src/notify.js
index c50ee1e..a3cd31a 100644
--- a/server/src/notify.js
+++ b/server/src/notify.js
@@ -193,6 +193,46 @@ export function sendFailure(type, url, failure, opts) {
}
}
+/**
+ * Pure: the one message sent when an "Update all" batch finishes.
+ *
+ * @param {{ succeeded: string[], failed: Array<{ name: string, message?: string }> }} summary
+ * @returns {{ title: string, body: string, ok: boolean }}
+ */
+export function buildBatchMessage({ succeeded = [], failed = [] }) {
+ const n = succeeded.length + failed.length;
+ const ok = failed.length === 0;
+ const title = ok
+ ? `✅ Update all finished: ${n} container${n === 1 ? '' : 's'} updated`
+ : `⚠️ Update all finished: ${succeeded.length} updated, ${failed.length} failed`;
+ const lines = [];
+ if (succeeded.length) lines.push(`Updated: ${succeeded.slice(0, MAX_LISTED).join(', ')}${succeeded.length > MAX_LISTED ? ', …' : ''}`);
+ for (const f of failed.slice(0, MAX_LISTED)) {
+ lines.push(`Failed: ${f.name}${f.message ? ` — ${String(f.message).split('\n')[0].slice(0, 200)}` : ''}`);
+ }
+ return { title, body: lines.join('\n'), ok };
+}
+
+/** Send the "Update all finished" summary to the configured target. */
+export function sendBatchSummary(type, url, summary, opts) {
+ const { title: t, body, ok } = buildBatchMessage(summary);
+ switch (type) {
+ case 'ntfy':
+ return postNtfy(url, { title: t, tags: ok ? 'white_check_mark' : 'warning', body }, opts);
+ case 'gotify':
+ return postJson(url, { title: t, message: body, priority: ok ? 4 : 7 }, opts);
+ case 'webhook':
+ return postJson(
+ url,
+ { title: t, message: body, event: 'batch_finished', succeeded: summary.succeeded, failed: summary.failed },
+ opts
+ );
+ case 'discord':
+ default:
+ return postJson(url, { content: `**${t}**\n${body}` }, opts);
+ }
+}
+
/** Send a one-off test message so the user can confirm their target works. */
export function sendTest(type, url, opts) {
const text = '✅ DockPull test — your notifications are configured correctly.';
@@ -222,4 +262,6 @@ export default {
sendTest,
sendFailure,
buildFailureMessage,
+ sendBatchSummary,
+ buildBatchMessage,
};
diff --git a/server/src/registry.js b/server/src/registry.js
index 3bb2232..6cbc6bf 100644
--- a/server/src/registry.js
+++ b/server/src/registry.js
@@ -149,15 +149,82 @@ export async function getRemoteDigest(imageRef, { timeoutMs = 10000 } = {}) {
* @param {Array<{ platform?: { os?: string, architecture?: string } }>} manifests
* @returns {object|null}
*/
-export function pickPlatformManifest(manifests) {
+export function pickPlatformManifest(manifests, arch = hostArch()) {
if (!Array.isArray(manifests) || manifests.length === 0) return null;
+ const real = manifests.filter((m) => m.platform?.os !== 'unknown'); // skip attestation entries
return (
- manifests.find((m) => m.platform?.os === 'linux' && m.platform?.architecture === 'amd64') ||
- manifests.find((m) => m.platform?.os === 'linux') ||
+ real.find((m) => m.platform?.os === 'linux' && m.platform?.architecture === arch) ||
+ real.find((m) => m.platform?.os === 'linux' && m.platform?.architecture === 'amd64') ||
+ real.find((m) => m.platform?.os === 'linux') ||
+ real[0] ||
manifests[0]
);
}
+/** This server's CPU as Docker names it (x64 -> amd64), so a Pi reads arm64 manifests. */
+export function hostArch() {
+ return { x64: 'amd64', arm64: 'arm64', arm: 'arm', ia32: '386', ppc64: 'ppc64le', s390x: 's390x' }[process.arch] || 'amd64';
+}
+
+/**
+ * Fetch a manifest (by tag or digest) with the registry's auth flow; follows
+ * an index/list to this host's platform. Returns the image manifest or null.
+ */
+async function fetchImageManifest(registry, repository, reference, timeoutMs, auth = { header: null }) {
+ const base = registryBaseUrl(registry);
+ const get = (ref) =>
+ fetch(`${base}/v2/${repository}/manifests/${encodeURIComponent(ref)}`, {
+ headers: { Accept: MANIFEST_ACCEPT, ...(auth.header ? { Authorization: auth.header } : {}) },
+ signal: AbortSignal.timeout(timeoutMs),
+ });
+ let res = await get(reference);
+ if (res.status === 401) {
+ auth.header = await resolveAuthHeader(res, registry, repository, timeoutMs);
+ if (auth.header) res = await get(reference);
+ }
+ if (!res.ok) return null;
+ let manifest = await res.json().catch(() => null);
+ if (Array.isArray(manifest?.manifests)) {
+ const picked = pickPlatformManifest(manifest.manifests);
+ if (!picked?.digest) return null;
+ const sub = await get(picked.digest);
+ if (!sub.ok) return null;
+ manifest = await sub.json().catch(() => null);
+ }
+ return manifest && Array.isArray(manifest.layers) ? manifest : null;
+}
+
+/**
+ * Pure: bytes to download to go from `oldLayers` to `newLayers` — Docker only
+ * pulls layers it doesn't already have. Sizes are the compressed (download)
+ * sizes from the manifests.
+ */
+export function newLayerBytes(newLayers, oldLayers = []) {
+ const have = new Set((oldLayers || []).map((l) => l.digest));
+ return (newLayers || []).reduce((sum, l) => (have.has(l.digest) ? sum : sum + (Number(l.size) || 0)), 0);
+}
+
+/**
+ * Best-effort: how much an update will download — the new image's layers that
+ * the running image doesn't already have. `exact` is false when the running
+ * image's manifest couldn't be read (then it's the whole new image, an upper
+ * bound). Never throws; null when unknown.
+ *
+ * @returns {Promise<{ bytes: number, exact: boolean }|null>}
+ */
+export async function getDownloadSize(imageRef, newDigest, oldDigest, { timeoutMs = 10000 } = {}) {
+ try {
+ const { registry, repository } = parseRef(imageRef);
+ const auth = { header: null };
+ const next = await fetchImageManifest(registry, repository, newDigest, timeoutMs, auth);
+ if (!next) return null;
+ const prev = oldDigest ? await fetchImageManifest(registry, repository, oldDigest, timeoutMs, auth) : null;
+ return { bytes: newLayerBytes(next.layers, prev?.layers), exact: Boolean(prev) };
+ } catch {
+ return null;
+ }
+}
+
async function authedJson(url, authHeader, timeoutMs) {
const res = await fetch(url, {
headers: {
@@ -285,4 +352,4 @@ export async function listTags(imageRef, { timeoutMs = 15000, maxPages = 10 } =
return tags;
}
-export default { registryBaseUrl, listTags, nextPageUrl, getRemoteDigest, getRemoteVersion, parseWwwAuthenticate, pickPlatformManifest };
+export default { registryBaseUrl, getDownloadSize, newLayerBytes, hostArch, listTags, nextPageUrl, getRemoteDigest, getRemoteVersion, parseWwwAuthenticate, pickPlatformManifest };
diff --git a/server/src/routes/api.js b/server/src/routes/api.js
index 700dbcb..fb5b11a 100644
--- a/server/src/routes/api.js
+++ b/server/src/routes/api.js
@@ -15,6 +15,7 @@ import {
listContainers,
getContainerImageMeta,
listDanglingImages,
+ getContainerLogs,
pruneDanglingImages,
removeDanglingImages,
shortImageId,
@@ -105,6 +106,7 @@ apiRouter.get('/api/containers', async (req, res) => {
getRollback: (name) => db.getRollbackPoint(name),
getCheckError: (ref) => errorByRef.get(ref) || null,
getTagUpdate: (ref) => db.getTagUpdate(ref),
+ getAutoRevert: (name) => db.getAutoRevert(name),
tagPolicy: getSettings().tagUpdates,
});
@@ -332,6 +334,26 @@ apiRouter.get('/api/self-update', async (req, res) => {
}
});
+// The container's own recent output (e.g. why an update crash-loops).
+apiRouter.get('/api/containers/:name/logs', async (req, res) => {
+ const tail = toSafeInt(req.query.tail, 200, 1000);
+ try {
+ res.set('Cache-Control', 'no-store');
+ return res.status(200).json(await getContainerLogs(req.params.name, tail));
+ } catch (err) {
+ if (err.statusCode === 404) return res.status(404).json({ error: 'not_found' });
+ return res.status(503).json({ error: 'docker_unavailable' });
+ }
+});
+
+// The saved live log of one past update (history rows carry `has_log`).
+apiRouter.get('/api/history/entry/:id/log', (req, res) => {
+ if (!/^\d{1,12}$/.test(req.params.id)) return res.status(400).json({ error: 'invalid_id' });
+ const log = db.getHistoryLog(Number(req.params.id));
+ if (log === null) return res.status(404).json({ error: 'not_found' });
+ return res.status(200).json({ log });
+});
+
// --- Backup / restore ---
apiRouter.get('/api/backup', (req, res) => {
diff --git a/server/src/routes/update.js b/server/src/routes/update.js
index 82edd95..3008da6 100644
--- a/server/src/routes/update.js
+++ b/server/src/routes/update.js
@@ -1,5 +1,8 @@
/**
- * Update routes: POST /api/update/:name, GET /api/update/:name/stream.
+ * Update routes. Every update, revert and tag switch is queued on the server
+ * (jobs.js) and runs there regardless of whether any page stays open; clients
+ * follow along via GET /api/jobs and the per-container live log stream
+ * GET /api/update/:name/stream (which replays the whole log on reconnect).
*
* Auth: protected by the session-cookie middleware mounted ahead of this
* router in index.js (see API_CONTRACT.md). This router adds no auth
@@ -14,190 +17,144 @@
*/
import express from 'express';
-import {
- docker,
- updateContainer,
- revertContainer,
- imageExists,
- trackedImageRef,
- switchContainerTag,
- restoreComposeEdit,
-} from '../docker.js';
+import { docker, imageExists, trackedImageRef, switchContainerTag } from '../docker.js';
import { normalizeRef } from '../reconcile.js';
import * as sse from '../sse.js';
+import * as jobs from '../jobs.js';
+import { executeUpdate, executeRevert } from '../update-runner.js';
import { getSettings } from '../settings.js';
-import { sendFailure } from '../notify.js';
+import { sendBatchSummary } from '../notify.js';
import * as db from '../db.js';
-import { validateContainerNameParam } from '../security.js';
+import { validateContainerNameParam, isValidContainerName } from '../security.js';
export const updateRouter = express.Router();
// Every :name route talks to Docker; reject anything that isn't a container name.
updateRouter.param('name', validateContainerNameParam);
-/**
- * Best-effort "update failed" notification. Never throws or delays the result.
- */
-function notifyFailure(failure) {
+// --- The queue → live clients and notifications -------------------------
+// Any change in the queue nudges dashboards (and the header's progress
+// indicator) to re-read GET /api/jobs; a finished job also changes containers.
+jobs.onChange((job) => {
+ sse.broadcastGlobal({ type: 'jobs-changed' });
+ if (job?.status === 'done') sse.broadcastGlobal({ type: 'containers-changed' });
+});
+
+jobs.onBatchDone((summary) => {
let s;
try {
s = getSettings();
} catch {
return;
}
- if (!s.discordEnabled || !s.discordWebhookUrl || !s.notifyOnFailure) return;
- sendFailure(s.notifyType, s.discordWebhookUrl, failure)
- .then((r) => {
- if (!r.ok) console.warn(`update.js: failure notification returned ${r.status}`);
- })
- .catch((err) => console.warn(`update.js: failure notification failed: ${err.message}`));
+ if (summary.total < 2 || !s.discordEnabled || !s.discordWebhookUrl || !s.notifyBatchSummary) return;
+ sendBatchSummary(s.notifyType, s.discordWebhookUrl, summary).catch((err) =>
+ console.warn(`update.js: batch summary notification failed: ${err.message}`)
+ );
+});
+
+const COMPOSE_PROJECT_LABEL = 'com.docker.compose.project';
+
+/** Inspect a container for queueing; maps Docker errors to HTTP responses. */
+async function inspectOr(res, name) {
+ try {
+ return await docker.getContainer(name).inspect();
+ } catch (err) {
+ if (err.statusCode === 404) res.status(404).json({ error: 'not_found' });
+ else if (err.code === 'ENOENT' || err.code === 'ECONNREFUSED') res.status(503).json({ error: 'docker_unavailable' });
+ else res.status(500).json({ error: 'internal_error', message: err.message });
+ return null;
+ }
}
/**
- * Runs the update + records history + finishes the SSE session, detached
- * from the request lifecycle (the POST handler responds before this
- * settles). Errors here must never escape as an unhandled rejection.
- *
- * @param {string} name
- * @param {string|null} image - configured image ref, for the history row.
+ * Queue an update for a container (shared by single updates and Update all).
+ * Starts its live log session right away so a client can attach while it waits.
*/
-async function runUpdate(name, image, { run = updateContainer } = {}) {
- try {
- const result = await run(name, (line, stream) => sse.pushLog(name, line, stream));
- const oldVersion = db.getImageVersion(result.oldDigest) ?? result.oldVersion ?? null;
- db.recordUpdate({
- container_name: name,
- image,
- old_digest: result.oldDigest,
- new_digest: result.newDigest,
- old_version: oldVersion,
- new_version: db.getImageVersion(result.newDigest),
- status: result.success ? 'success' : 'failure',
- message: result.message,
- });
- // On success, clear any pending update event for this image so the
- // dashboard indicator goes away — we just pulled the latest. Relying on
- // the digest-equality check in /api/containers alone is not enough:
- // A registry can report a manifest-list (multi-arch) digest while the
- // container's RepoDigest is platform-specific, so they'd never match and
- // the badge would stick forever.
- if (result.success && image) {
- try {
- db.resolveEventsForRef(normalizeRef(image));
- } catch {
- // normalizeRef shouldn't throw for a real image ref; non-fatal.
- }
- }
- // Remember how to undo this update (the previous local image) whenever the
- // image actually changed — even on a health-downgraded "failure", so the
- // user can revert a broken update.
- // Compare local image IDs, not registry digests: the old image's digest
- // may be unknown (untagged by an earlier update of a sibling container).
- if (result.oldImageId && result.newImageId && result.oldImageId !== result.newImageId) {
- db.setRollbackPoint({
- container_name: name,
- image_id: result.oldImageId,
- image_ref: image,
- old_digest: result.oldDigest,
- old_version: oldVersion,
- compose_edit: result.composeEdit ?? null,
- });
- }
- sse.finish(name, { success: result.success, message: result.message });
- if (!result.success) notifyFailure({ name, image, action: 'update', message: result.message });
- } catch (err) {
- db.recordUpdate({
- container_name: name,
- image,
- old_digest: null,
- new_digest: null,
- status: 'failure',
- message: err.message,
- });
- sse.finish(name, { success: false, message: err.message });
- notifyFailure({ name, image, action: 'update', message: err.message });
- } finally {
- // Let other connected dashboards refresh their list/badges.
- sse.broadcastGlobal({ type: 'containers-changed' });
+function queueUpdate(name, inspectData, { batchId = null, run, kind = 'update', meta = {} } = {}) {
+ const image = trackedImageRef(inspectData);
+ const project = inspectData.Config?.Labels?.[COMPOSE_PROJECT_LABEL] || null;
+ const job = jobs.enqueue({
+ name,
+ kind,
+ project,
+ batchId,
+ meta,
+ execute: () => executeUpdate(name, image, run),
+ });
+ sse.startSession(name);
+ if (job.status === 'queued') {
+ sse.pushLog(
+ name,
+ job.waitingOn === 'stack'
+ ? `Queued — updates in stack "${project}" run one at a time…`
+ : 'Queued — waiting for a free slot (up to 3 updates run at once)…',
+ 'stdout'
+ );
}
+ return job;
+}
+
+function busy(res) {
+ return res.status(409).json({ error: 'update_in_progress' });
}
updateRouter.post('/api/update/:name', async (req, res) => {
const { name } = req.params;
+ const inspectData = await inspectOr(res, name);
+ if (!inspectData) return undefined;
+ if (jobs.isBusy(name)) return busy(res);
+ const job = queueUpdate(name, inspectData);
+ return res.status(200).json({ streamId: name, job });
+});
- let inspectData;
- try {
- inspectData = await docker.getContainer(name).inspect();
- } catch (err) {
- if (err.statusCode === 404) {
- return res.status(404).json({ error: 'not_found' });
+/**
+ * "Update all": queue every named container in one batch, run entirely on the
+ * server — closing the page or locking the phone no longer stops it.
+ * Body: { names: string[] }. Containers that are busy or gone are reported in
+ * `skipped`, never silently dropped.
+ */
+updateRouter.post('/api/update-all', async (req, res) => {
+ const names = Array.isArray(req.body?.names) ? [...new Set(req.body.names)] : null;
+ if (!names || names.length === 0 || names.length > 500 || !names.every(isValidContainerName)) {
+ return res.status(400).json({ error: 'invalid_payload' });
+ }
+ const batchId = jobs.startBatch();
+ const queued = [];
+ const skipped = [];
+ for (const name of names) {
+ if (jobs.isBusy(name)) {
+ skipped.push({ name, reason: 'already updating' });
+ continue;
}
- if (err.code === 'ENOENT' || err.code === 'ECONNREFUSED') {
- return res.status(503).json({ error: 'docker_unavailable' });
+ let inspectData;
+ try {
+ inspectData = await docker.getContainer(name).inspect();
+ } catch (err) {
+ skipped.push({ name, reason: err.statusCode === 404 ? 'not found' : 'docker unavailable' });
+ continue;
}
- return res.status(500).json({ error: 'internal_error', message: err.message });
+ queueUpdate(name, inspectData, { batchId });
+ queued.push(name);
}
+ jobs.sealBatch(batchId);
+ return res.status(200).json({ batchId, queued, skipped });
+});
- if (sse.isActive(name)) {
- return res.status(409).json({ error: 'update_in_progress' });
- }
-
- sse.startSession(name);
-
- const image = trackedImageRef(inspectData);
- // Fire-and-forget: don't await, so the POST returns promptly. runUpdate
- // catches its own errors, so this can never reject/crash the process.
- void runUpdate(name, image);
-
- return res.status(200).json({ streamId: name });
+/** What's queued/running now (and what finished recently), for any client. */
+updateRouter.get('/api/jobs', (req, res) => {
+ res.set('Cache-Control', 'no-store');
+ return res.status(200).json(jobs.snapshot());
});
-/**
- * Detached revert: recreate the container from its remembered previous image,
- * record history, and finish the SSE session. Mirrors runUpdate.
- */
-async function runRevert(name, image, rollback) {
- try {
- // Undo a tag switch's compose-file edit too, so the file and the container
- // agree (and a later `compose up` doesn't move it forward again).
- if (rollback.compose_edit) {
- restoreComposeEdit(rollback.compose_edit, (l) => sse.pushLog(name, l, 'stdout'));
- }
- const result = await revertContainer(name, rollback.image_id, (line, stream) => sse.pushLog(name, line, stream), {
- imageRef: image,
- digest: rollback.old_digest,
- });
- db.recordUpdate({
- container_name: name,
- image,
- old_digest: result.oldDigest,
- new_digest: result.newDigest,
- old_version: db.getImageVersion(result.oldDigest),
- new_version: rollback.old_version ?? null,
- status: result.success ? 'success' : 'failure',
- message: result.message,
- });
- // Consume the rollback point on a successful revert (can't revert twice to
- // the same image). The update it reverted away from will simply be
- // re-detected as available on the next check.
- if (result.success) db.deleteRollbackPoint(name);
- sse.finish(name, { success: result.success, message: result.message });
- if (!result.success) notifyFailure({ name, image, action: 'revert', message: result.message });
- } catch (err) {
- db.recordUpdate({
- container_name: name,
- image,
- old_digest: null,
- new_digest: null,
- status: 'failure',
- message: err.message,
- });
- sse.finish(name, { success: false, message: err.message });
- notifyFailure({ name, image, action: 'revert', message: err.message });
- } finally {
- sse.broadcastGlobal({ type: 'containers-changed' });
- }
-}
+/** Per-container "revert automatically if the update comes up unhealthy". */
+updateRouter.put('/api/containers/:name/auto-revert', (req, res) => {
+ const enabled = req.body?.enabled;
+ if (typeof enabled !== 'boolean') return res.status(400).json({ error: 'invalid_payload' });
+ db.setAutoRevert(req.params.name, enabled);
+ sse.broadcastGlobal({ type: 'containers-changed' });
+ return res.status(200).json({ ok: true, enabled });
+});
updateRouter.post('/api/update/:name/revert', async (req, res) => {
const { name } = req.params;
@@ -206,24 +163,12 @@ updateRouter.post('/api/update/:name/revert', async (req, res) => {
if (!rollback) {
return res.status(404).json({ error: 'no_rollback' });
}
-
- let inspectData;
- try {
- inspectData = await docker.getContainer(name).inspect();
- } catch (err) {
- if (err.statusCode === 404) return res.status(404).json({ error: 'not_found' });
- if (err.code === 'ENOENT' || err.code === 'ECONNREFUSED') {
- return res.status(503).json({ error: 'docker_unavailable' });
- }
- return res.status(500).json({ error: 'internal_error' });
- }
-
- if (sse.isActive(name)) {
- return res.status(409).json({ error: 'update_in_progress' });
- }
+ const inspectData = await inspectOr(res, name);
+ if (!inspectData) return undefined;
+ if (jobs.isBusy(name)) return busy(res);
// The saved image can disappear (pruned, or removed by hand). Check before
- // touching the container, and forget the dead rollback point.
+ // queueing, and forget the dead rollback point.
try {
if (!(await imageExists(rollback.image_id))) {
db.deleteRollbackPoint(name);
@@ -237,11 +182,16 @@ updateRouter.post('/api/update/:name/revert', async (req, res) => {
return res.status(503).json({ error: 'docker_unavailable' });
}
- sse.startSession(name);
const image = rollback.image_ref ?? trackedImageRef(inspectData);
- void runRevert(name, image, rollback);
-
- return res.status(200).json({ streamId: name });
+ const project = inspectData.Config?.Labels?.[COMPOSE_PROJECT_LABEL] || null;
+ const job = jobs.enqueue({
+ name,
+ kind: 'revert',
+ project,
+ execute: () => executeRevert(name, image, rollback),
+ });
+ sse.startSession(name);
+ return res.status(200).json({ streamId: name, job });
});
// Docker tag grammar: [A-Za-z0-9_][A-Za-z0-9_.-]{0,127}
@@ -258,13 +208,8 @@ updateRouter.post('/api/update/:name/switch-tag', async (req, res) => {
return res.status(400).json({ error: 'invalid_tag' });
}
- let inspectData;
- try {
- inspectData = await docker.getContainer(name).inspect();
- } catch (err) {
- if (err.statusCode === 404) return res.status(404).json({ error: 'not_found' });
- return res.status(503).json({ error: 'docker_unavailable' });
- }
+ const inspectData = await inspectOr(res, name);
+ if (!inspectData) return undefined;
const image = trackedImageRef(inspectData);
let ref;
try {
@@ -280,15 +225,16 @@ updateRouter.post('/api/update/:name/switch-tag', async (req, res) => {
});
}
- if (sse.isActive(name)) {
- return res.status(409).json({ error: 'update_in_progress' });
- }
- sse.startSession(name);
+ if (jobs.isBusy(name)) return busy(res);
// The tag_updates row belongs to the OLD tag and may still apply to other
// containers running it, so it's left alone; the switched container now has
// a new ref, whose newer tags the next check works out.
- void runUpdate(name, image, { run: (n, onLine) => switchContainerTag(n, tag, onLine) });
- return res.status(200).json({ streamId: name });
+ const job = queueUpdate(name, inspectData, {
+ kind: 'switch',
+ meta: { tag },
+ run: (n, onLine) => switchContainerTag(n, tag, onLine),
+ });
+ return res.status(200).json({ streamId: name, job });
});
updateRouter.get('/api/update/:name/stream', (req, res) => {
diff --git a/server/src/settings.js b/server/src/settings.js
index 5f1508e..e731c53 100644
--- a/server/src/settings.js
+++ b/server/src/settings.js
@@ -96,6 +96,12 @@ const SPEC = {
fromStore: (v) => bool(v, true),
fromInput: (v) => (typeof v === 'boolean' ? v : undefined),
},
+ // Send one summary message when an "Update all" batch finishes.
+ notifyBatchSummary: {
+ default: true,
+ fromStore: (v) => bool(v, true),
+ fromInput: (v) => (typeof v === 'boolean' ? v : undefined),
+ },
// Master "send notifications" toggle (kept this key for back-compat).
discordEnabled: {
default: ENV_WEBHOOK !== '',
diff --git a/server/src/sse.js b/server/src/sse.js
index c8b637d..833763a 100644
--- a/server/src/sse.js
+++ b/server/src/sse.js
@@ -16,7 +16,11 @@
const sessions = new Map();
-const FINISHED_SESSION_TTL_MS = 30_000;
+// Keep a finished update's log long enough for a dashboard that was closed or
+// on another page to come back and still see what happened.
+const FINISHED_SESSION_TTL_MS = 15 * 60_000;
+// Cap per-session memory (a pull of a big image prints a lot of progress).
+const MAX_LINES = 2000;
/**
* @param {string} name
@@ -73,9 +77,40 @@ export function pushLog(name, line, stream) {
const evt = { type: 'log', line, stream };
session.lines.push(evt);
+ if (session.lines.length > MAX_LINES) {
+ session.lines.splice(0, session.lines.length - MAX_LINES);
+ session.dropped = (session.dropped || 0) + 1;
+ }
+ session.total = (session.total || 0) + 1;
writeToSubscribers(session, evt);
}
+/**
+ * How many log lines a session has produced so far (including any trimmed
+ * from memory) — a cursor for logTail(name, n, since).
+ */
+export function logCount(name) {
+ return sessions.get(name)?.total || 0;
+}
+
+/**
+ * The last `n` log lines of a session as text, optionally only those produced
+ * after cursor `since` (from logCount). Used to save an update's log with its
+ * history entry.
+ */
+export function logTail(name, n = 200, since = 0) {
+ const session = sessions.get(name);
+ if (!session) return '';
+ const total = session.total || 0;
+ const firstKept = total - session.lines.length;
+ const from = Math.max(0, since - firstKept);
+ return session.lines
+ .slice(from)
+ .slice(-n)
+ .map((e) => e.line)
+ .join('\n');
+}
+
/**
* Marks the session as finished: records the final result, writes it to
* every subscriber, ends their responses, and schedules the session for
@@ -246,6 +281,8 @@ export function broadcastGlobal(evt) {
}
export default {
+ logCount,
+ logTail,
startSession,
isActive,
pushLog,
diff --git a/server/src/update-runner.js b/server/src/update-runner.js
new file mode 100644
index 0000000..8a16324
--- /dev/null
+++ b/server/src/update-runner.js
@@ -0,0 +1,183 @@
+/**
+ * The work behind each queued job (see jobs.js): run an update / tag switch /
+ * revert, record history (including the tail of the live log), remember or
+ * consume the rollback point, and notify on failure.
+ *
+ * Opt-in auto-revert: when an update or tag switch succeeds at pulling and
+ * recreating but the container then comes up unhealthy, and the container has
+ * "revert automatically" enabled, the previous image is restored straight away
+ * in the same job (its log continues in the same live stream).
+ *
+ * Each `execute*` returns `{ success, message }` and finishes the container's
+ * live log session (sse.finish) with that final outcome.
+ */
+
+import { updateContainer, revertContainer, imageExists, restoreComposeEdit } from './docker.js';
+import { normalizeRef } from './reconcile.js';
+import * as sse from './sse.js';
+import { getSettings } from './settings.js';
+import { sendFailure } from './notify.js';
+import * as db from './db.js';
+
+const LOG_LINES_SAVED = 200;
+
+/** Best-effort "update failed" notification. Never throws or delays the result. */
+export function notifyFailure(failure) {
+ let s;
+ try {
+ s = getSettings();
+ } catch {
+ return;
+ }
+ if (!s.discordEnabled || !s.discordWebhookUrl || !s.notifyOnFailure) return;
+ sendFailure(s.notifyType, s.discordWebhookUrl, failure)
+ .then((r) => {
+ if (!r.ok) console.warn(`update-runner: failure notification returned ${r.status}`);
+ })
+ .catch((err) => console.warn(`update-runner: failure notification failed: ${err.message}`));
+}
+
+const logFor = (name) => (line, stream) => sse.pushLog(name, line, stream);
+
+/**
+ * Update (or tag switch, via `run`) one container and record the outcome.
+ * Does NOT finish the live session — the caller may continue it (auto-revert).
+ *
+ * @returns {Promise<{ success: boolean, message: string, unhealthy: boolean }>}
+ */
+async function performUpdate(name, image, run) {
+ const since = sse.logCount(name);
+ try {
+ const result = await run(name, logFor(name));
+ const oldVersion = db.getImageVersion(result.oldDigest) ?? result.oldVersion ?? null;
+ db.recordUpdate({
+ container_name: name,
+ image,
+ old_digest: result.oldDigest,
+ new_digest: result.newDigest,
+ old_version: oldVersion,
+ new_version: db.getImageVersion(result.newDigest),
+ status: result.success ? 'success' : 'failure',
+ message: result.message,
+ log: sse.logTail(name, LOG_LINES_SAVED, since),
+ });
+ // On success, clear any pending update event for this image so the
+ // dashboard indicator goes away (a registry index digest and a platform
+ // RepoDigest may never compare equal, so don't rely on that alone).
+ if (result.success && image) {
+ try {
+ db.resolveEventsForRef(normalizeRef(image));
+ } catch {
+ // non-fatal
+ }
+ }
+ // Remember how to undo this update whenever the image actually changed —
+ // even on a health-downgraded "failure", so a broken update can be
+ // reverted. Compare local image IDs: the old digest may be unknown.
+ if (result.oldImageId && result.newImageId && result.oldImageId !== result.newImageId) {
+ db.setRollbackPoint({
+ container_name: name,
+ image_id: result.oldImageId,
+ image_ref: image,
+ old_digest: result.oldDigest,
+ old_version: oldVersion,
+ compose_edit: result.composeEdit ?? null,
+ });
+ }
+ return { success: result.success, message: result.message, unhealthy: result.healthy === false };
+ } catch (err) {
+ db.recordUpdate({
+ container_name: name,
+ image,
+ status: 'failure',
+ message: err.message,
+ log: sse.logTail(name, LOG_LINES_SAVED, since),
+ });
+ return { success: false, message: err.message, unhealthy: false };
+ }
+}
+
+/**
+ * Revert a container to its rollback point and record the outcome. Does NOT
+ * finish the live session.
+ *
+ * @returns {Promise<{ success: boolean, message: string }>}
+ */
+async function performRevert(name, image, rollback) {
+ const since = sse.logCount(name);
+ try {
+ if (!(await imageExists(rollback.image_id))) {
+ db.deleteRollbackPoint(name);
+ const message = 'The previous image no longer exists (it may have been pruned), so this update can no longer be reverted.';
+ sse.pushLog(name, message, 'stderr');
+ return { success: false, message };
+ }
+ // Undo a tag switch's compose-file edit too, so the file and the container agree.
+ if (rollback.compose_edit) restoreComposeEdit(rollback.compose_edit, (l) => sse.pushLog(name, l, 'stdout'));
+ const result = await revertContainer(name, rollback.image_id, logFor(name), {
+ imageRef: image,
+ digest: rollback.old_digest,
+ });
+ db.recordUpdate({
+ container_name: name,
+ image,
+ old_digest: result.oldDigest,
+ new_digest: result.newDigest,
+ old_version: db.getImageVersion(result.oldDigest),
+ new_version: rollback.old_version ?? null,
+ status: result.success ? 'success' : 'failure',
+ message: result.message,
+ log: sse.logTail(name, LOG_LINES_SAVED, since),
+ });
+ // A rollback point is single-use; the update it undid is re-detected later.
+ if (result.success) db.deleteRollbackPoint(name);
+ return { success: result.success, message: result.message };
+ } catch (err) {
+ db.recordUpdate({
+ container_name: name,
+ image,
+ status: 'failure',
+ message: err.message,
+ log: sse.logTail(name, LOG_LINES_SAVED, since),
+ });
+ return { success: false, message: err.message };
+ }
+}
+
+/**
+ * Job body for an update or tag switch (`run` defaults to a plain update).
+ * Applies opt-in auto-revert on an unhealthy result.
+ */
+export async function executeUpdate(name, image, run = updateContainer) {
+ let outcome = await performUpdate(name, image, run);
+ if (!outcome.success && outcome.unhealthy && db.getAutoRevert(name)) {
+ const rollback = db.getRollbackPoint(name);
+ if (rollback) {
+ sse.pushLog(name, '— Unhealthy after updating; reverting automatically (auto-revert is on). —', 'stdout');
+ const reverted = await performRevert(name, rollback.image_ref ?? image, rollback);
+ // "Image updated, but . The new version may be broken — …" ->
+ // just the reason; the advice to revert no longer applies.
+ const reason = /but (.+?)\. The new version/.exec(outcome.message)?.[1] || 'it came up unhealthy';
+ const prev = rollback.old_version ? `version ${rollback.old_version}` : 'the previous image';
+ outcome = reverted.success
+ ? { success: false, message: `The update didn't work (${reason}), so it was automatically reverted to ${prev}.` }
+ : {
+ success: false,
+ message: `The update didn't work (${reason}), and the automatic revert also failed: ${reverted.message}`,
+ };
+ }
+ }
+ sse.finish(name, outcome);
+ if (!outcome.success) notifyFailure({ name, image, action: 'update', message: outcome.message });
+ return outcome;
+}
+
+/** Job body for a revert. */
+export async function executeRevert(name, image, rollback) {
+ const outcome = await performRevert(name, image, rollback);
+ sse.finish(name, outcome);
+ if (!outcome.success) notifyFailure({ name, image, action: 'revert', message: outcome.message });
+ return outcome;
+}
+
+export default { executeUpdate, executeRevert, notifyFailure };
diff --git a/server/test/checker.test.js b/server/test/checker.test.js
index 078c0df..f329f10 100644
--- a/server/test/checker.test.js
+++ b/server/test/checker.test.js
@@ -15,14 +15,19 @@ process.env.DATA_DIR = tmp;
const D = (c) => `sha256:${c.repeat(64)}`;
// --- fake registry ------------------------------------------------------
-const registry = { digests: {}, tags: {}, hits: [] }; // digests["app:1.0.0"] = digest
+const registry = { digests: {}, tags: {}, manifests: {}, hits: [] }; // digests["app:1.0.0"] = digest
const regServer = http.createServer((req, res) => {
registry.hits.push(`${req.method} ${req.url}`);
const m = /^\/v2\/(.+)\/(manifests|tags)\/(.+?)(\?.*)?$/.exec(req.url);
if (!m) return res.writeHead(404).end();
const [, repo, kind, rest] = m;
if (kind === 'manifests') {
- const digest = registry.digests[`${repo}:${decodeURIComponent(rest)}`];
+ const ref = decodeURIComponent(rest);
+ if (ref.startsWith('sha256:') && registry.manifests[ref]) {
+ res.writeHead(200, { 'Content-Type': 'application/json' });
+ return res.end(JSON.stringify(registry.manifests[ref]));
+ }
+ const digest = registry.digests[`${repo}:${ref}`];
if (!digest) return res.writeHead(404).end();
res.writeHead(200, { 'Docker-Content-Digest': digest, 'Content-Type': 'application/json' });
return res.end(req.method === 'HEAD' ? undefined : '{}');
@@ -81,6 +86,7 @@ function reset() {
docker.images = {};
registry.digests = {};
registry.tags = {};
+ registry.manifests = {};
registry.hits = [];
db.db.exec('DELETE FROM update_events; DELETE FROM tag_updates;');
}
@@ -162,3 +168,18 @@ test('concurrent checks share one run (no duplicate registry calls or events)',
const n = db.db.prepare('SELECT COUNT(*) AS n FROM update_events WHERE resolved = 0').get().n;
assert.equal(n, 1);
});
+
+test('records how much an update will download (only layers not already present)', async () => {
+ reset();
+ const ref = `${REG}/app:latest`;
+ addImage(D('1'), { repoDigests: [`${REG}/app@${D('a')}`], tagRef: ref });
+ docker.containers.web = { image: ref, imageId: D('1') };
+ registry.digests['app:latest'] = D('f');
+ const base = { digest: D('9'), size: 50_000_000 };
+ registry.manifests[D('a')] = { layers: [base, { digest: D('7'), size: 3_000_000 }] };
+ registry.manifests[D('f')] = { layers: [base, { digest: D('8'), size: 4_000_000 }] };
+ await runCheck();
+ const ev = db.latestUnresolvedEventForRef(`${REG}/app:latest`);
+ assert.equal(ev.download_bytes, 4_000_000);
+ assert.equal(ev.download_exact, 1);
+});
diff --git a/server/test/containers-service.test.js b/server/test/containers-service.test.js
index 364a261..d15929c 100644
--- a/server/test/containers-service.test.js
+++ b/server/test/containers-service.test.js
@@ -159,10 +159,12 @@ describe('buildContainerItems', () => {
availableVersion: null,
breakingRisk: false,
skipped: false,
+ downloadSize: null,
newerTag: null,
newerMajorTag: null,
pinned: false,
canRevert: false,
+ autoRevert: false,
rollbackVersion: null,
checkError: null,
state: 'running',
diff --git a/server/test/docker.test.js b/server/test/docker.test.js
index bc36213..ed5f393 100644
--- a/server/test/docker.test.js
+++ b/server/test/docker.test.js
@@ -173,3 +173,39 @@ test('healthWaitPlan: waits as long as the container\'s own healthcheck needs',
assert.equal(healthWaitPlan({ Test: ['CMD', 'x'], StartPeriod: 3600 * s }).timeoutMs, 900_000);
assert.equal(healthWaitPlan({ Test: ['CMD', 'x'], Interval: 1 * s, Timeout: 1 * s, Retries: 1 }).timeoutMs, 30_000);
});
+
+import { demuxDockerLogs } from '../src/docker.js';
+
+test('demuxDockerLogs: strips frame headers from stdout/stderr frames', () => {
+ const frame = (stream, text) => {
+ const payload = Buffer.from(text);
+ const h = Buffer.alloc(8);
+ h[0] = stream;
+ h.writeUInt32BE(payload.length, 4);
+ return Buffer.concat([h, payload]);
+ };
+ const buf = Buffer.concat([frame(1, 'hello\n'), frame(2, 'oops\n'), frame(1, 'bye\n')]);
+ assert.equal(demuxDockerLogs(buf), 'hello\noops\nbye\n');
+});
+
+test('demuxDockerLogs: raw TTY text passes through unchanged', () => {
+ assert.equal(demuxDockerLogs(Buffer.from('plain tty output\n')), 'plain tty output\n');
+ assert.equal(demuxDockerLogs(Buffer.alloc(0)), '');
+});
+
+import { mapLimit } from '../src/docker.js';
+
+test('mapLimit: preserves order and never exceeds the limit', async () => {
+ let active = 0;
+ let max = 0;
+ const out = await mapLimit([5, 1, 4, 2, 3, 0], 3, async (n) => {
+ active += 1;
+ max = Math.max(max, active);
+ await new Promise((r) => setTimeout(r, n * 3));
+ active -= 1;
+ return n * 10;
+ });
+ assert.deepEqual(out, [50, 10, 40, 20, 30, 0]);
+ assert.equal(max, 3);
+ assert.deepEqual(await mapLimit([], 3, async () => 1), []);
+});
diff --git a/server/test/jobs.test.js b/server/test/jobs.test.js
new file mode 100644
index 0000000..53cd7d8
--- /dev/null
+++ b/server/test/jobs.test.js
@@ -0,0 +1,102 @@
+import { test, beforeEach } from 'node:test';
+import assert from 'node:assert/strict';
+import * as jobs from '../src/jobs.js';
+
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+
+// A fake job that records when it ran, so ordering/overlap can be asserted.
+function recorder() {
+ const events = [];
+ let active = 0;
+ let maxActive = 0;
+ const make = (name, ms = 20, success = true) => async () => {
+ active += 1;
+ maxActive = Math.max(maxActive, active);
+ events.push(`start ${name}`);
+ await sleep(ms);
+ events.push(`end ${name}`);
+ active -= 1;
+ return { success, message: success ? 'ok' : 'boom' };
+ };
+ return { events, make, get maxActive() { return maxActive; } };
+}
+
+beforeEach(async () => {
+ await jobs._drain();
+ jobs._reset();
+});
+
+test('same stack runs one at a time, in order', async () => {
+ const r = recorder();
+ jobs.enqueue({ name: 'a', kind: 'update', project: 'web', execute: r.make('a') });
+ jobs.enqueue({ name: 'b', kind: 'update', project: 'web', execute: r.make('b') });
+ await jobs._drain();
+ assert.deepEqual(r.events, ['start a', 'end a', 'start b', 'end b']);
+});
+
+test('different stacks run in parallel, capped at 3', async () => {
+ const r = recorder();
+ for (const n of ['a', 'b', 'c', 'd', 'e']) jobs.enqueue({ name: n, kind: 'update', project: n, execute: r.make(n, 30) });
+ await jobs._drain();
+ assert.equal(r.maxActive, 3);
+ assert.equal(r.events.filter((e) => e.startsWith('end')).length, 5);
+});
+
+test('one job per container; a second is refused while busy', async () => {
+ const r = recorder();
+ jobs.enqueue({ name: 'a', kind: 'update', execute: r.make('a') });
+ assert.equal(jobs.isBusy('a'), true);
+ assert.throws(() => jobs.enqueue({ name: 'a', kind: 'revert', execute: r.make('a2') }), { code: 'busy' });
+ await jobs._drain();
+ assert.equal(jobs.isBusy('a'), false);
+ jobs.enqueue({ name: 'a', kind: 'revert', execute: r.make('a3') }); // fine once done
+ await jobs._drain();
+});
+
+test('a throwing job is recorded as failed and does not block its lane', async () => {
+ const r = recorder();
+ jobs.enqueue({ name: 'x', kind: 'update', project: 'p', execute: async () => { throw new Error('kaboom'); } });
+ jobs.enqueue({ name: 'y', kind: 'update', project: 'p', execute: r.make('y') });
+ await jobs._drain();
+ const snap = jobs.snapshot();
+ const x = snap.jobs.find((j) => j.name === 'x');
+ assert.equal(x.success, false);
+ assert.equal(x.message, 'kaboom');
+ assert.equal(snap.jobs.find((j) => j.name === 'y').success, true);
+});
+
+test('batch: summary reported once, when the last job finishes', async () => {
+ const r = recorder();
+ const done = [];
+ jobs.onBatchDone((s) => done.push(s));
+ const id = jobs.startBatch();
+ jobs.enqueue({ name: 'a', kind: 'update', project: 's1', batchId: id, execute: r.make('a') });
+ jobs.enqueue({ name: 'b', kind: 'update', project: 's1', batchId: id, execute: r.make('b', 20, false) });
+ jobs.enqueue({ name: 'c', kind: 'update', project: 's2', batchId: id, execute: r.make('c') });
+ jobs.sealBatch(id);
+ assert.equal(jobs.snapshot().batch.total, 3);
+ await jobs._drain();
+ assert.equal(done.length, 1);
+ assert.deepEqual(done[0].succeeded.sort(), ['a', 'c']);
+ assert.deepEqual(done[0].failed.map((f) => f.name), ['b']);
+ assert.ok(jobs.snapshot().batch.finishedAt);
+});
+
+test('snapshot keeps finished jobs for a while, then forgets them', async () => {
+ const r = recorder();
+ jobs.enqueue({ name: 'a', kind: 'update', execute: r.make('a') });
+ await jobs._drain();
+ assert.equal(jobs.snapshot().jobs.length, 1);
+ assert.equal(jobs.snapshot(Date.now() + 16 * 60_000).jobs.length, 0);
+});
+
+test('a queued job says why it waits: its stack, or the parallel cap', async () => {
+ const r = recorder();
+ for (const n of ['a', 'b', 'c']) jobs.enqueue({ name: n, kind: 'update', project: n, execute: r.make(n, 30) });
+ const sameStack = jobs.enqueue({ name: 'a2', kind: 'update', project: 'a', execute: r.make('a2') });
+ assert.equal(sameStack.waitingOn, 'stack');
+ await new Promise((res) => setTimeout(res, 5)); // let a, b, c take the 3 slots
+ const capped = jobs.enqueue({ name: 'd', kind: 'update', project: 'd', execute: r.make('d') });
+ assert.equal(capped.waitingOn, 'slot');
+ await jobs._drain();
+});
diff --git a/server/test/notify.test.js b/server/test/notify.test.js
index da7f97a..0a082a5 100644
--- a/server/test/notify.test.js
+++ b/server/test/notify.test.js
@@ -131,3 +131,15 @@ test('sendFailure: every target type delivers (ntfy emoji title via query)', asy
server.close();
}
});
+
+import { buildBatchMessage } from '../src/notify.js';
+
+test('buildBatchMessage: all good vs some failed', () => {
+ const ok = buildBatchMessage({ succeeded: ['a', 'b'], failed: [] });
+ assert.match(ok.title, /2 containers updated/);
+ assert.equal(ok.ok, true);
+ const mixed = buildBatchMessage({ succeeded: ['a'], failed: [{ name: 'db', message: 'Image updated, but it keeps restarting\nmore' }] });
+ assert.match(mixed.title, /1 updated, 1 failed/);
+ assert.match(mixed.body, /Failed: db — Image updated, but it keeps restarting$/m);
+ assert.doesNotMatch(mixed.body, /more/);
+});
diff --git a/server/test/registry.test.js b/server/test/registry.test.js
index 8c13f18..383864f 100644
--- a/server/test/registry.test.js
+++ b/server/test/registry.test.js
@@ -69,3 +69,26 @@ test('nextPageUrl: follows same-origin rel=next links only', () => {
assert.equal(nextPageUrl(null, base), null);
assert.equal(nextPageUrl('; rel="prev"', base), null);
});
+
+import { newLayerBytes, hostArch } from '../src/registry.js';
+
+test('newLayerBytes: only layers the old image lacks count', () => {
+ const a = { digest: 'sha256:a', size: 100 };
+ const b = { digest: 'sha256:b', size: 20 };
+ const c = { digest: 'sha256:c', size: 5 };
+ assert.equal(newLayerBytes([a, b, c], [a, b]), 5);
+ assert.equal(newLayerBytes([a, c]), 105); // unknown old image -> whole image
+ assert.equal(newLayerBytes([], [a]), 0);
+});
+
+test('pickPlatformManifest: prefers this host\'s CPU and skips attestation entries', () => {
+ const manifests = [
+ { digest: 'sha256:att', platform: { os: 'unknown', architecture: 'unknown' } },
+ { digest: 'sha256:amd', platform: { os: 'linux', architecture: 'amd64' } },
+ { digest: 'sha256:arm', platform: { os: 'linux', architecture: 'arm64' } },
+ ];
+ assert.equal(pickPlatformManifest(manifests, 'arm64').digest, 'sha256:arm');
+ assert.equal(pickPlatformManifest(manifests, 'amd64').digest, 'sha256:amd');
+ assert.equal(pickPlatformManifest(manifests, 'riscv64').digest, 'sha256:amd');
+ assert.ok(['amd64', 'arm64', 'arm', '386', 'ppc64le', 's390x'].includes(hostArch()));
+});
diff --git a/server/test/settings.test.js b/server/test/settings.test.js
index 0fc6c89..0b1da63 100644
--- a/server/test/settings.test.js
+++ b/server/test/settings.test.js
@@ -20,6 +20,7 @@ test('settings: defaults when nothing stored', () => {
scheduleMode: 'daily',
scheduleIntervalHours: 6,
notifyOnFailure: true,
+ notifyBatchSummary: true,
discordEnabled: false,
discordWebhookUrl: '',
tagUpdates: 'minor',
diff --git a/server/test/sse.test.js b/server/test/sse.test.js
index 4e3f345..070b840 100644
--- a/server/test/sse.test.js
+++ b/server/test/sse.test.js
@@ -50,7 +50,7 @@ test('a finished session is still replayable during the grace period, then remov
sse.subscribe('db', late);
assert.match(late.chunks.join(''), /pulling[\s\S]*"success":true/);
- mock.timers.tick(31_000);
+ mock.timers.tick(15 * 60_000 + 1000);
const later = fakeRes();
sse.subscribe('db', later);
assert.match(later.chunks.join(''), /No active update/);
@@ -58,3 +58,21 @@ test('a finished session is still replayable during the grace period, then remov
mock.timers.reset();
}
});
+
+test('logTail: last N lines, optionally only those after a cursor', () => {
+ mock.timers.enable({ apis: ['setTimeout'] });
+ try {
+ sse.startSession('tail');
+ sse.pushLog('tail', 'update line 1');
+ sse.pushLog('tail', 'update line 2');
+ const cursor = sse.logCount('tail');
+ sse.pushLog('tail', 'revert line 1');
+ assert.equal(sse.logTail('tail', 200), 'update line 1\nupdate line 2\nrevert line 1');
+ assert.equal(sse.logTail('tail', 1), 'revert line 1');
+ assert.equal(sse.logTail('tail', 200, cursor), 'revert line 1');
+ assert.equal(sse.logTail('nope'), '');
+ sse.finish('tail', { success: true, message: 'ok' });
+ } finally {
+ mock.timers.reset();
+ }
+});