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')