From 29e0cc8f6e3155accedc5230a9811ba9dfdfe728 Mon Sep 17 00:00:00 2001 From: Daan Meijer Date: Mon, 26 Oct 2026 22:53:52 +0100 Subject: [PATCH] The window worker keeps its window over a chunk that doesn't inflate A live reply cut short (the server's compressed data ending at ".\r\n" at a read boundary; the spool keeps what Corpus.write refuses) made the worker give the group back without a snapshot, losing the window and the batch since its last post (six times in the first 40 minutes on homelab). Now the batch before the chunk goes out, the chunk is reported (fetched again, plain the second time) and the range is asked for again from that chunk (RangeOutcome::Again): the stream waits for the new copy and the window stays. Test: the coordinator hands the same range back at its resume point after a corrupt report. Co-Authored-By: Claude Fable 5.1 --- REBUILD.md | 6 +++++- edge/README.md | 4 +++- edge/src/bin/edge-node.rs | 23 ++++++++++++++++++----- tests/functional/group_leases.spec.ts | 9 ++++++++- 4 files changed, 34 insertions(+), 8 deletions(-) diff --git a/REBUILD.md b/REBUILD.md index 6f9a891..034c7a8 100644 --- a/REBUILD.md +++ b/REBUILD.md @@ -539,7 +539,11 @@ releases, a third of them one poster's image dump (27.7 M rows, LOG.md default (`--ranges` keeps the old path). Tests: the lease order, takeover and redo, the endpoints with a corpus-served stream and the snapshot through the API, the snapshot restoring a window between two - ranges (the two-range release one, the album joined). + ranges (the two-range release one, the album joined). Deployed and + started 2026-10-26 22:37 (LOG.md (118)). A chunk that doesn't inflate + (a live reply cut short; the spool doesn't check) is reported after + the batch before it went out and the range asked for again from that + chunk: the window stays (the first run lost it, and the open batch). 6. **The corpus stays as it is.** Measured on 300 unfiltered chunks (30 M lines, LOG.md (102)): 43% of lines lack a "(k/N)" counter, and of diff --git a/edge/README.md b/edge/README.md index b87b9e4..8b135d3 100644 --- a/edge/README.md +++ b/edge/README.md @@ -256,7 +256,9 @@ COORDINATOR_URL=http://coordinator:3334 NODE_TOKEN=… ./target/release/edge-nod 409 means the lease is lost (the range stops, isn't given back); a batch once 500k articles or 100k files wait (a picture group's batches went over the server's 256 MB limit); a chunk that doesn't inflate is - reported (the coordinator fetches it plain next time); SIGTERM/SIGINT + reported (the coordinator fetches it plain next time) after the batch + before it went out, and a held group asks for the range again from + that chunk, its window kept; SIGTERM/SIGINT give each range back after its current chunk; `--max-mbit` is the node's download cap, shared by its threads. Plus a heartbeat when no batch went out for a minute. diff --git a/edge/src/bin/edge-node.rs b/edge/src/bin/edge-node.rs index 3b07e66..cf0b1e1 100644 --- a/edge/src/bin/edge-node.rs +++ b/edge/src/bin/edge-node.rs @@ -536,6 +536,9 @@ fn run_range(api: &Arc, lease: LeaseJson, config: &Config, stopping: &Atomi /// What one range of a held group came to. enum RangeOutcome { Done, + /// A chunk didn't inflate: reported (the coordinator fetches it again, plain the second time) after the batch + /// before it went out, and the range is asked for again from there; the window stays. + Again, /// The group is given back (a stop, a lost lease, a failure): no more ranges. Stop, } @@ -626,17 +629,25 @@ fn run_group_range(api: &Arc, group_id: i64, group_name: &str, range: Range } } } - // Stopped mid-range: the batch so far goes out, so the snapshot stands at a stored resume point. - if !reached_end && corrupt.is_none() && failure.is_none() && post_error.lock().unwrap().is_none() && indexer.next() > batch_from { + // Stopped mid-range (or at a chunk that doesn't inflate): the batch so far goes out, so the snapshot stands at + // a stored resume point, and a chunk reported corrupt is read again from its own start. + if !reached_end && failure.is_none() && post_error.lock().unwrap().is_none() && indexer.next() > batch_from { let _ = post_tx.send(indexer.take()); } drop(post_tx); let _ = poster.join(); let post_error = post_error.lock().unwrap().take(); if let Some((from, to, e)) = corrupt { - log("warn", &format!("{group_name} {}–{}: chunk {from}–{to} doesn't inflate ({e}); reported, the group goes back", range.start, range.end)); - let _ = api.corrupt(range.id, from); - return RangeOutcome::Stop; + // The server's compressed reply cut short (LOG.md 2026-10-11): the coordinator fetches the chunk again, + // plain the second time, and sets the range aside after that; the window carries on either way. + log("warn", &format!("{group_name} {}–{}: chunk {from}–{to} doesn't inflate ({e}); reported, the range is asked for again", range.start, range.end)); + if let Err(e) = api.corrupt(range.id, from) { + log("warn", &format!("{group_name} {}–{}: reporting the chunk: {e}; the group goes back", range.start, range.end)); + return RangeOutcome::Stop; + } + if post_error.is_none() { + return RangeOutcome::Again; + } } match post_error { Some(ApiError::LeaseLost(m)) => { @@ -758,6 +769,8 @@ fn run_group(api: &Arc, lease: GroupLeaseJson, config: &Config, stopping: & } match run_group_range(api, group_id, &group_name, range, &mut indexer, config, stopping, &totals) { RangeOutcome::Done => ranges += 1, + // The same range again from the reported chunk: the stream waits for its new copy. + RangeOutcome::Again => {} RangeOutcome::Stop => break, } if snapshotted.elapsed() >= config.snapshot_every { diff --git a/tests/functional/group_leases.spec.ts b/tests/functional/group_leases.spec.ts index 6b9b8ce..b82de71 100644 --- a/tests/functional/group_leases.spec.ts +++ b/tests/functional/group_leases.spec.ts @@ -6,7 +6,7 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' import coordinatorConfig from '#config/coordinator' import GroupLeases from '#services/group_leases' -import { LeaseLost } from '#services/leases' +import Leases, { LeaseLost } from '#services/leases' import { Readable } from 'node:stream' async function node(name: string) { @@ -71,6 +71,13 @@ test.group('Group leases', (g) => { assert.equal((await GroupLeases.next(b, n1))?.range.id, b1) await db.from('ranges').where('id', b1).update({ status: 'done', next_id: 101 }) assert.equal((await GroupLeases.next(b, n1))?.range.id, b2) + // A chunk of b2 that doesn't inflate, reported after the batch before it: the range is asked for again and + // comes back at its resume point (the chunk fetched plain next time). + await db.from('ranges').where('id', b2).update({ next_id: 151 }) + await Leases.corrupt(b2, n1, 151) + assert.include(await rangeRow(b2), { status: 'planned', node_id: null }) + assert.deepEqual((await GroupLeases.next(b, n1))?.range, { id: b2, start: 101, end: 200, next: 151 }) + assert.deepEqual((await rangeRow(b2)).plain_chunks.map(Number), [151]) await db.from('ranges').where('id', b2).update({ status: 'done', next_id: 201 }) assert.isNull(await GroupLeases.next(b, n1), 'caught up')