Edge Queues and Scheduled Jobs

The previous lesson managed to hold coordinated state at the edge: the single-writer object queued up writes arriving for its own key. Asynchronous work, though, still runs in the center; validation, invoice lines, and field notifications get written to a queue in the center region. This lesson runs the same job set in two places and counts three things separately: the queue-entry trip, where the job runs, and the result’s return path.

The measurement runs on the six branch edges of the fictional measurement network and is a model: the edges, the two queue services, and the scheduler are set up with node. No real product is called.

EC25. For every edge, the one-way latency to the center and to the nearest edge location is given, along with a nightly read count. EC26. The queue-entry trip is twice the one-way latency plus the queue’s fixed acceptance time. EC27. The center queue offers a batch read of 250, 10 retries, 14 days of retention, and ordering per key; the edge queue offers 25 and 3, retains for 4 days, and gives no ordering. EC28. The consumer starts once per batch read: 40 ms in the center, 6 ms at the edge. Processing per read is 12 ms on both paths. EC29. The result is written to the store where the job runs. The center store is as far away as the center trip; the edge key-value store is read locally, but its write takes 480 ms to become visible at other edges. EC30. Scheduled triggering knows only the base clock; the edge scheduler chooses the trigger location itself, and the choice is not recorded.

The Same Job Set, Two Queues

// network.mjs — fictional measurement network model: edges, latencies, the two queues' constraints.
export const EDGES = [                              // one-way latency ms, band offset from base clock
  { name: "edge-west",     center: 12, edge: 5,  band: 0, reads: 610 },
  { name: "edge-north",    center: 38, edge: 7,  band: 0, reads: 520 },
  { name: "edge-east",     center: 74, edge: 9,  band: 2, reads: 430 },
  { name: "edge-south",    center: 51, edge: 6,  band: 1, reads: 350 },
  { name: "edge-highland", center: 96, edge: 11, band: 3, reads: 280 },
  { name: "edge-port",     center: 63, edge: 8,  band: 2, reads: 210 },
];
export const TOTAL = EDGES.reduce((t, u) => t + u.reads, 0);
export const QUEUE = {                              // constraints, item by item
  center: { accept: 9, batch: 250, retries: 10, retentionDays: 14, startup: 40 },
  edge:   { accept: 3, batch: 25,  retries: 3,  retentionDays: 4,  startup: 6 },
};
export const PROCESSING = 12, PROPAGATION = 480, STORE_READ = 5;   // processing / edge propagation / store response, ms
export const roundTrip = (path, u) => 2 * u[path] + QUEUE[path].accept;
export const row = (...s) => console.log(s.map((x, i) => String(x).padEnd(i ? 13 : 15)).join("").trimEnd());

export function rng(seed) {                          // seed is visible; same sequence run to run
  let s = seed >>> 0;
  return () => (s = (s * 1664525 + 1013904223) >>> 0) / 4294967296;
}
export function jobSet(seed) {                        // reading: counter id, increasing sequence, required retries
  const r = rng(seed), seq = new Map(), items = [];
  for (const u of EDGES) for (let i = 0; i < u.reads; i++) {
    const counterId = `s${String(Math.floor(r() * 900)).padStart(3, "0")}`;
    seq.set(counterId, (seq.get(counterId) ?? 0) + 1);
    items.push({ site: u.name, counterId, seq: seq.get(counterId), retries: 1 });
  }
  for (const i of items) while (r() < 0.28 && i.retries < 12) i.retries++;
  return items;
}

// trip.mjs — the same job set in two places: queue-entry trip, where the job runs, the result's return path.
import { EDGES, TOTAL, QUEUE, PROCESSING, PROPAGATION, STORE_READ, roundTrip, row } from "./network.mjs";

row("edge", "reads", "center trip", "edge trip", "gain");
for (const u of EDGES) row(u.name, u.reads, `${roundTrip("center", u)} ms`, `${roundTrip("edge", u)} ms`,
  `${(roundTrip("center", u) / roundTrip("edge", u)).toFixed(1)}x`);
const weighted = (path) => EDGES.reduce((t, u) => t + u.reads * roundTrip(path, u), 0) / TOTAL;
console.log(`weighted queue-entry trip for ${TOTAL} reads: ` +
  `center ${weighted("center").toFixed(1)} ms, edge ${weighted("edge").toFixed(1)} ms\n`);

for (const path of ["center", "edge"]) {            // consumer: how many times it starts
  const k = QUEUE[path], calls = Math.ceil(TOTAL / k.batch);
  const startup = calls * k.startup, work = TOTAL * PROCESSING;
  console.log(`${path.padEnd(7)}batch read ${String(k.batch).padStart(3)} -> ${String(calls).padStart(3)} consumer calls; ` +
    `startup ${String(startup).padStart(4)} ms + processing ${work} ms = ${startup + work} ms`);
}
console.log(`where the job runs: on the center path, a single known region; ` +
  `on the edge path, a location the platform chooses, and the choice is not recorded\n`);

// return path: the center store is remote, the edge store is local but must propagate first
const returnTrip = (path) => EDGES.reduce((t, u) => t + u.reads * (2 * u[path] + STORE_READ), 0) / TOTAL;
console.log(`weighted return trip: center ${returnTrip("center").toFixed(1)} ms, edge ${returnTrip("edge").toFixed(1)} ms`);
console.log(`until the result is ready: center ${returnTrip("center").toFixed(1)} ms, ` +
  `edge ${(returnTrip("edge") + PROPAGATION).toFixed(1)} ms (propagation ${PROPAGATION} ms added)`);
console.log(`end-to-end total: center ${(weighted("center") + returnTrip("center")).toFixed(1)} ms, ` +
  `edge ${(weighted("edge") + returnTrip("edge") + PROPAGATION).toFixed(1)} ms`);

edge           reads        center trip  edge trip    gain
edge-west      610          33 ms        13 ms        2.5x
edge-north     520          85 ms        17 ms        5.0x
edge-east      430          157 ms       21 ms        7.5x
edge-south     350          111 ms       15 ms        7.4x
edge-highland  280          201 ms       25 ms        8.0x
edge-port      210          135 ms       19 ms        7.1x
weighted queue-entry trip for 2400 reads: center 106.4 ms, edge 17.5 ms

center batch read 250 ->  10 consumer calls; startup  400 ms + processing 28800 ms = 29200 ms
edge   batch read  25 ->  96 consumer calls; startup  576 ms + processing 28800 ms = 29376 ms
where the job runs: on the center path, a single known region; on the edge path, a location the platform chooses, and the choice is not recorded

weighted return trip: center 102.4 ms, edge 19.5 ms
until the result is ready: center 102.4 ms, edge 499.5 ms (propagation 480 ms added)
end-to-end total: center 208.8 ms, edge 517.0 ms

On the first measure, the edge wins by a wide margin: the weighted queue-entry trip drops from 106.4 ms to 17.5 ms, and at the farthest edge the gain reaches eightfold.

On the second measure, the table turns around. The edge consumer starts in six milliseconds, the center in forty; but because the edge’s batch size is twenty-five, 2,400 messages split into ninety-six calls, against ten in the center. A startup that is cheaper per call ends up costlier in total because it starts nine times as often. Where the job runs also becomes uncertain: a single known region in the center, an unrecorded location the platform chooses at the edge.

The third measure is the return path. The edge’s return trip is short at 19.5 ms, but because the result is written to the edge store, becoming visible at the user’s edge waits on the 480 ms propagation: the end-to-end total comes to 208.8 ms in the center, 517.0 ms at the edge. The 89 ms gained on queue entry does not make up for the 397 ms lost in making the result readable.

The Edge Queue’s Four Constraints

// constraint.mjs — the edge queue's four constraints counted on the same job set: retries, retention, ordering, batch.
import { TOTAL, QUEUE, jobSet, rng } from "./network.mjs";

const row = (...s) => console.log(s.map((x, i) => String(x).padEnd(i ? 15 : 17)).join("").trimEnd());
const items = jobSet(20260731);                     // seed is visible
row("constraint", "center", "edge", "measure", "edge's loss");

const deadLettered = (k) => items.filter((i) => i.retries > k.retries).length;   // messages that exceed the retry budget
row("retries", QUEUE.center.retries, QUEUE.edge.retries,
  `${TOTAL} messages`, `${deadLettered(QUEUE.edge)} dead letters (${(100 * deadLettered(QUEUE.edge) / TOTAL).toFixed(1)}%)`);

const OUTAGE_DAYS = 6, DAILY = TOTAL;               // consumer stopped for six days
const retained = (k) => Math.min(OUTAGE_DAYS, k.retentionDays) * DAILY;
row("retention", `${QUEUE.center.retentionDays} d`, `${QUEUE.edge.retentionDays} d`,
  `${OUTAGE_DAYS} d outage`, `${OUTAGE_DAYS * DAILY - retained(QUEUE.edge)} messages dropped`);

const r = rng(770419), shuffled = items.map((i) => [r(), i]).sort((a, b) => a[0] - b[0]).map((x) => x[1]);
const apply = (stream) => { const final = new Map(); for (const i of stream) final.set(i.counterId, i.seq); return final; };
const correct = apply([...items].sort((a, b) => a.seq - b.seq)), actual = apply(shuffled);
const wrong = [...correct].filter(([s, n]) => actual.get(s) !== n).length;
row("ordering", "per key", "none", `${correct.size} counters`,
  `${wrong} counters with wrong index (${(100 * wrong / correct.size).toFixed(1)}%)`);

const calls = (k) => Math.ceil(TOTAL / k.batch);
row("batch read", QUEUE.center.batch, QUEUE.edge.batch, `${TOTAL} messages`,
  `${calls(QUEUE.edge) - calls(QUEUE.center)} extra calls`);

const multiRead = [...correct.values()].filter((n) => n > 1).length, edgeQ = QUEUE.edge;
console.log("\nthe cost of working around it, item by item:");
for (const s of [
  `retries: ${deadLettered(edgeQ)} dead letters requeued by hand -> ${deadLettered(edgeQ)} extra writes; the retention clock resets, sequence starts over`,
  `retention: a second durable copy for ${OUTAGE_DAYS * DAILY - retained(edgeQ)} messages; all ${OUTAGE_DAYS * DAILY} messages get written to two places`,
  `ordering: a buffer for ${multiRead} multi-read counters; ${TOTAL} extra state writes, a 3000 ms wait per message`,
  `batch read: ${calls(edgeQ)} x ${edgeQ.startup} ms = ${calls(edgeQ) * edgeQ.startup} ms, more than the center's ${calls(QUEUE.center) * QUEUE.center.startup} ms`,
]) console.log("  " + s);

constraint       center         edge           measure        edge's loss
retries          10             3              2400 messages  50 dead letters (2.1%)
retention        14 d           4 d            6 d outage     4800 messages dropped
ordering         per key        none           842 counters   438 counters with wrong index (52.0%)
batch read       250            25             2400 messages  86 extra calls

the cost of working around it, item by item:
  retries: 50 dead letters requeued by hand -> 50 extra writes; the retention clock resets, sequence starts over
  retention: a second durable copy for 4800 messages; all 14400 messages get written to two places
  ordering: a buffer for 667 multi-read counters; 2400 extra state writes, a 3000 ms wait per message
  batch read: 96 x 6 ms = 576 ms, more than the center's 400 ms

The four constraints yield four separate numbers. When the retry allowance drops from ten to three, 50 messages fall into the dead letter queue under the same error distribution; on the center path, all of them get through. When retention drops from fourteen days to four, of the 14,400 messages that pile up during a six-day outage, 4,800 fall out of the queue; the center queue loses none.

The most expensive item is ordering. Without the guarantee, 438 of 842 counters end up with the wrong index: the last read applied may not be the highest sequence number, and an invoice comes out short or over. This is not an item a setting can fix — it is a guarantee the queue never offered in the first place.

The cost of working around it sits in the last block. Requeuing dead letters by hand means 50 extra writes, but the requeued message is new: its retention clock resets and its sequence number starts over. Longer retention can only be had with a second durable copy, and ordering only with a buffer kept in the name of 667 counters: all 14,400 messages get written to two places, producing 2,400 extra state writes and a three-second wait per message. All three are larger than the 89 ms gained at the edge.

Which Region a Scheduled Job Triggers In

// schedule.mjs — the scheduled job: which region it fires in, and who owns the time-zone conversion.
import { EDGES, STORE_READ } from "./network.mjs";

const BASE = 2, DRIFT_DAYS = 217, DRIFTING_BANDS = [2, 3];   // base clock 02:00, seasonal drift window
const hour = (h) => `${String(((h % 24) + 24) % 24).padStart(2, "0")}:00`;
const dayNote = (h) => (h < 0 ? " (previous day)" : "");

const row = (...s) => console.log(s.map((x, i) => String(x).padEnd(i ? 16 : 15)).join("").trimEnd());
row("edge", "band", "single trigger", "local time", "drift window");
for (const u of EDGES) row(u.name, `+${u.band}`, hour(BASE), hour(BASE + u.band),
  DRIFTING_BANDS.includes(u.band) ? `${DRIFT_DAYS} d 60 min` : "none");

const localHours = new Set(EDGES.map((u) => BASE + u.band));
const bands = [...new Set(EDGES.map((u) => u.band))].sort();
console.log(`\nsingle trigger -> ${localHours.size} distinct local times; ` +
  `${EDGES.filter((u) => u.band === 0).length}/${EDGES.length} edges see the requested ${hour(BASE)}`);
console.log(`every edge needs ${bands.length} separate trigger definitions for ${hour(BASE)}: ` +
  bands.map((s) => `+${s} base ${hour(BASE - s)}${dayNote(BASE - s)}`).join(", "));
const drifting = EDGES.filter((u) => DRIFTING_BANDS.includes(u.band));
console.log(`seasonal drift: ${drifting.length} edges x ${DRIFT_DAYS} days = ${drifting.length * DRIFT_DAYS} edge-days ` +
  `of 60 min deviation; the scheduler knows only the base clock, the function does the conversion itself\n`);

const READS = 2400, PACKET = 100, rounds = Math.ceil(READS / PACKET);
const duration = (u) => rounds * (2 * u.center + STORE_READ);
const l = EDGES.map(duration), min = Math.min(...l), max = Math.max(...l);
console.log(`nightly job: ${rounds} rounds x ${PACKET} reads; because the trigger location cannot be chosen, duration ` +
  `falls between ${min}-${max} ms, average ${(l.reduce((a, b) => a + b) / l.length).toFixed(0)} ms ` +
  `(${(max / min).toFixed(1)}x spread)`);
console.log(`the cost of pinning: if the job is pinned to the center region, duration is fixed at ${min} ms, ` +
  `but the ability to run at the edge is given up entirely for this job`);

edge           band            single trigger  local time      drift window
edge-west      +0              02:00           02:00           none
edge-north     +0              02:00           02:00           none
edge-east      +2              02:00           04:00           217 d 60 min
edge-south     +1              02:00           03:00           none
edge-highland  +3              02:00           05:00           217 d 60 min
edge-port      +2              02:00           04:00           217 d 60 min

single trigger -> 4 distinct local times; 2/6 edges see the requested 02:00
every edge needs 4 separate trigger definitions for 02:00: +0 base 02:00, +1 base 01:00, +2 base 00:00, +3 base 23:00 (previous day)
seasonal drift: 3 edges x 217 days = 651 edge-days of 60 min deviation; the scheduler knows only the base clock, the function does the conversion itself

nightly job: 24 rounds x 100 reads; because the trigger location cannot be chosen, duration falls between 696-4728 ms, average 2792 ms (6.8x spread)
the cost of pinning: if the job is pinned to the center region, duration is fixed at 696 ms, but the ability to run at the edge is given up entirely for this job

The nightly job’s request to “run at 02:00 at every branch” cannot be satisfied by a single trigger definition. A trigger set on the base clock lands on four distinct local times, and only two of the six edges see the requested hour. Every edge needs four definitions for 02:00, and band +3’s falls on base 23:00 — that is, the previous day. Two bands also drift sixty minutes on 217 days of the year: 651 edge-days of deviation. The scheduler knows only the base clock; the decision to convert to local time is not handed over, it stays in the function’s code.

The last two lines count where the trigger actually runs. The nightly job reads the center store in twenty-four rounds; because where it runs cannot be chosen, the same job finishes anywhere between 696 and 4,728 ms, averaging 2,792 ms — a 6.8-fold range, and which edge it lands on is not our decision. The only way to fix the duration is to pin the job to the center; then it does not run at the edge.

Summary

• The queue-entry trip fell at the edge from 106.4 ms to 17.5 ms, and the gain at the farthest edge reached 8-fold; but because the batch read dropped from 250 to 25, 2,400 messages split into 96 calls instead of 10, and total startup came to 576 ms instead of 400.
• Because the result was written to the edge store, it waited on a 480 ms propagation; the end-to-end total came to 208.8 ms in the center, 517.0 ms at the edge. Acceptance sped up, delivery slowed down.
• The four constraints yielded four numbers: a 3-retry allowance produced 50 dead letters, 4-day retention lost 4,800 messages in a 6-day outage, the absence of an ordering guarantee left 438 of 842 counters with the wrong index, and a batch of 25 cost 86 extra calls. Working around them cost 50 extra writes, writing 14,400 messages to two places, 2,400 extra state writes, and a 3,000 ms wait.
• A single trigger fell across 4 distinct local times, and 2 of 6 edges saw the requested hour; seasonal drift came to 651 edge-days, and the nightly job took 696-4,728 ms depending on where it was triggered.

Next Step

The edge runtime reached a queue and a scheduler, but how that access gets set up was never asked. A resource’s address and credentials can be handed over by building a client at runtime, or by a declarative binding at deploy time. The next lesson reaches the same resource both ways and counts four things: lines of code, startup work, where the credential sits, and the moment a misconfiguration is noticed.