Twigpine/node
1
0 Fork 0
mirror of https://github.com/Twigpine/node.git synced 2026-11-09 05:20:05 -10:00
Code Issues Packages Projects Releases 1 Wiki Activity

fix(node): mint a continuation when a ceiling stops the /ipfs scan mid-page

Browse Source 
The probe and visit ceilings taint inside gate_and_serve and returned Skip, so
the row they refused and every row behind it were walked past without a verdict
while the resume position was sealed from pager.cursor, the end of the fetched
page. Two ways that stranded content:

- On the final page, `pager.exhausted` breaks ahead of every mint arm, so the
  shed carried no continuation at all. A tokenless search_incomplete is the
  wrapped-scan answer, which tells `gl ipfs get` its ladder is over, so a holder
  on that page was unreachable on every retry.
- Mid-page, the sealed cursor sat past the refused rows, so the resume skipped
  them. The shipped ladder tests all set page_rows == ceiling, which puts the
  break exactly on a page boundary and hides both.

The ceiling now returns GateOutcome::CeilingStop rather than tainting on its way
out, and the scan loop stops there and seals the row in front of the one that was
refused. record_scan_truncation is the single site that records a truncation:
it taints and seals together, and only ever moves the position forward, so a
later oid candidate re-walking the same rows on a spent budget cannot hand back a
token the caller already echoed. The wrapped-scan tail no longer clears a seal,
since a ceiling can stop a resumed scan part way through the last page.
This commit is contained in:
beardthelion committed 2026-08-15 18:38:09 -05:00
1 parent a7e8e35030
commit 69cfbd4b3c
1 file changed
+328 -15
Download Patch File Download Diff File Expand all files Collapse all files
crates/gitlawb-node/src/api/ipfs.rs
+328 -15
View File
@@ -824,6 +824,14 @@ pub async fn get_by_cid(
throttled = true;
continue;
}
// The provenance path targets a bounded source list rather than the
// table, so there is no scan position to resume from: taint and move on,
// exactly as before. Only the visit ceiling can reach here (the probe
// ceiling is `legacy_scan`-only).
GateOutcome::CeilingStop(reason) => {
record_scan_truncation(&mut walk, &mut scan_continuation, reason, None);
continue;
}
GateOutcome::Skip => continue,
}
}
@@ -958,13 +966,21 @@ pub async fn get_by_cid(
// byte-identical to the wrapped-scan answer that tells the caller
// their ladder is over.
if walk.probes >= state.ipfs_max_legacy_probes {
walk.taint("probe-ceiling");
scan_continuation = pager.cursor.clone();
record_scan_truncation(
&mut walk,
&mut scan_continuation,
"probe-ceiling",
pager.cursor.clone(),
);
break;
}
if walk.visits >= state.config.ipfs_max_repo_visits {
walk.taint("visit-ceiling");
scan_continuation = pager.cursor.clone();
record_scan_truncation(
&mut walk,
&mut scan_continuation,
"visit-ceiling",
pager.cursor.clone(),
);
break;
}
// Row ceiling (F2). The two checks above only bind once a probe or a
@@ -974,8 +990,12 @@ pub async fn get_by_cid(
// probes, anonymously, while holding a scarce walk permit. This is
// the check that actually stops that scan.
if pager.fetched_rows >= state.ipfs_max_legacy_scan_rows {
walk.taint("row-ceiling");
scan_continuation = pager.cursor.clone();
record_scan_truncation(
&mut walk,
&mut scan_continuation,
"row-ceiling",
pager.cursor.clone(),
);
break;
}
// Rule-bytes ceiling: the row ceiling bounds rows, not the rules each
@@ -984,8 +1004,12 @@ pub async fn get_by_cid(
// tail is never materialized; `fetch_next_page` drops the rows behind
// it and the request that asked for them is the one that truncates.
if pager.rule_bytes_exceeded {
walk.taint("rules-ceiling");
scan_continuation = pager.cursor.clone();
record_scan_truncation(
&mut walk,
&mut scan_continuation,
"rules-ceiling",
pager.cursor.clone(),
);
break;
}
// Page toll (F2). Every page is work bought by an anonymous caller,
@@ -1038,6 +1062,32 @@ pub async fn get_by_cid(
// A throttled walk-requiring candidate is skipped, not fatal:
// keep scanning for a later walk-free copy (#173 review, F-C).
GateOutcome::Throttled => throttled = true,
// A ceiling refused THIS row, so the resume position is the row in
// front of it, not `pager.cursor`, which by now sits at the end of
// the fetched page and would skip every row the ceiling refused. The
// four arms at the top of the loop seal `pager.cursor` legitimately:
// they only fire once every fetched row has been walked.
//
// Stopping here rather than skipping on is also why the FINAL page is
// covered: the `pager.exhausted` break sits ahead of every mint arm,
// so a ceiling reached while walking the last page used to shed with
// no token at all.
GateOutcome::CeilingStop(reason) => {
// Nothing in front of it means nothing was settled, so there is
// no position to seal and this stop contributes none. The first
// oid candidate cannot get here (the ceiling arms above run
// before every fetch, so a spent budget breaks there instead);
// a LATER candidate can, because it re-walks the already-fetched
// rows from the front without passing those arms. Sealing
// `pager.cursor` for it would push the resume position past rows
// that candidate never examined.
let resume = (idx >= 2).then(|| {
let prev = &pager.rows[idx - 2];
(prev.created_at_key.clone(), prev.repo.id.clone())
});
record_scan_truncation(&mut walk, &mut scan_continuation, reason, resume);
break;
}
GateOutcome::Skip => {}
}
}
@@ -1058,9 +1108,13 @@ pub async fn get_by_cid(
//
// A wrapped scan emits NO continuation: there is nothing left to resume, and the
// absence of the token is what tells the caller their ladder is over.
if pager.resumed && pager.exhausted {
//
// Gated on nothing having been sealed: a ceiling can stop a resumed scan PART WAY
// through the last page, which leaves `exhausted` set with rows still unwalked in
// front of the cursor. Clearing the seal there would strand exactly those rows,
// which is the same tokenless dead end this clause exists to describe honestly.
if pager.resumed && pager.exhausted && scan_continuation.is_none() {
walk.taint("scan-wrapped");
scan_continuation = None;
}
// Nothing served — four distinct tails, in precedence order:
@@ -1151,6 +1205,36 @@ enum GateOutcome {
/// A walk-requiring candidate hit the per-IP walk quota; skip it but let the caller
/// record the throttle so a later walk-free copy can still serve.
Throttled,
/// A per-request CEILING (probes, repo visits) refused this row before it could reach
/// a verdict, and will refuse every row after it too. Distinct from `Skip` because the
/// caller must both taint AND seal a resume position in front of this row: the taint
/// alone sheds a 503 whose missing token reads as "ladder over", stranding this row
/// and everything behind it on an inventory that never changes.
CeilingStop(&'static str),
}
/// The one site that records a scan truncation: taint the walk with the reason and seal
/// the position the caller echoes back, together.
///
/// Keeping the two together is the point. Every earlier drip on this path was a ceiling
/// that tainted somewhere the mint could not see, so the shed carried no token.
///
/// The position only ever moves FORWARD. A later oid candidate re-walks the same fetched
/// rows from the front with the request's budget already spent, so it stops earlier than
/// the candidate before it; letting that overwrite the seal would hand back a token the
/// caller already echoed and the ladder would never advance.
fn record_scan_truncation(
walk: &mut WalkState,
slot: &mut Option<(String, String)>,
reason: &'static str,
pos: Option<(String, String)>,
) {
walk.taint(reason);
if let Some(pos) = pos {
if slot.as_ref().is_none_or(|sealed| pos > *sealed) {
*slot = Some(pos);
}
}
}
/// Outcome of the bounded, off-worker object read for one gated candidate (F6, #173).
@@ -1323,9 +1407,10 @@ async fn gate_and_serve(
if legacy_scan {
if walk.probes >= state.ipfs_max_legacy_probes {
// Budget spent: stop probing and mark the scan truncated so the tail
// reports an incomplete search (503), not a false 404 (#173, F2).
walk.taint("probe-ceiling");
return GateOutcome::Skip;
// reports an incomplete search (503), not a false 404 (#173, F2). The
// CALLER records it, because the resume position belongs to the row this
// refused and only the caller knows it.
return GateOutcome::CeilingStop("probe-ceiling");
}
if let Some(key) =
crate::rate_limit::client_key(ctx.headers, ctx.peer, state.push_limiter_trust)
@@ -1346,8 +1431,7 @@ async fn gate_and_serve(
"/ipfs request hit the per-request repo-visit ceiling \
(GITLAWB_IPFS_MAX_REPO_VISITS); skipping repo without a verdict"
);
walk.taint("visit-ceiling");
return GateOutcome::Skip;
return GateOutcome::CeilingStop("visit-ceiling");
}
walk.visits += 1;
@@ -5005,6 +5089,235 @@ mod tests {
);
}
/// The probe ceiling must ladder past the row it STOPPED ON, not past the page.
///
/// The sibling test above sets `ipfs_legacy_scan_page_rows == ipfs_max_legacy_probes`,
/// so the budget runs out exactly at a page boundary and the page-boundary cursor
/// happens to be the right resume point. Misalign the two and it is not: the ceiling
/// taints INSIDE `gate_and_serve`, the loop keeps consuming the rest of the page as
/// `Skip`, and the mint arms at the top of the loop seal `pager.cursor`, which by
/// then sits PAST every row the ceiling refused to probe. Those rows are skipped on
/// the resume as well, and the inventory is stable, so every ladder step reproduces
/// the same gap.
///
/// Two rows, one probe: the filler spends the budget, the holder is the row the
/// ceiling stops on.
#[cfg(unix)]
#[sqlx::test]
async fn get_by_cid_probe_ceiling_ladders_past_the_row_it_stopped_on(pool: sqlx::PgPool) {
let tmp = tempfile::TempDir::new().unwrap();
let mut state = crate::test_support::test_state(pool.clone()).await;
let repos_dir = tmp.path().join("repos");
std::fs::create_dir_all(&repos_dir).unwrap();
state.repo_store = crate::git::repo_store::RepoStore::for_testing(repos_dir, pool.clone());
state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
state.ipfs_legacy_scan_page_rows = 2;
state.ipfs_max_legacy_scan_rows = 1024;
// Deliberately NOT equal to the page size: one probe, two rows per page.
state.ipfs_max_legacy_probes = 1;
state.ipfs_work_rate_limiter =
crate::rate_limit::RateLimiter::new(1024, std::time::Duration::from_secs(3600));
seed_root_readable_repos(&state, "midpage", 1).await;
stamp_scan_order(&pool, "z6readablemidpage/midpage-0000", 0).await;
let (holder_id, oid) = seed_repo_with_blob(
&state,
tmp.path(),
"z6midpage",
"holder",
b"stopped on this row\n",
)
.await;
stamp_scan_order(&pool, &holder_id, 1).await;
let cid = seed_legacy_pin_for_oid(&state, &oid).await;
let router = ipfs_router(state);
let peer: SocketAddr = "203.0.113.161:5000".parse().unwrap();
let mut token: Option<String> = None;
let mut served_at = None;
for step in 1..=4 {
let (status, body) = status_and_body(
router
.clone()
.oneshot(get_cid_scan(&cid, Some(peer), token.as_deref()))
.await
.unwrap(),
)
.await;
if status == StatusCode::OK {
served_at = Some(step);
break;
}
assert_eq!(
status,
StatusCode::SERVICE_UNAVAILABLE,
"an intermediate rung is the retryable 503 (step {step}): {body}"
);
token = Some(continuation_of(&body).unwrap_or_else(|| {
panic!("the probe-ceiling shed at step {step} must carry a continuation: {body}")
}));
}
assert!(
served_at.is_some(),
"the row the probe ceiling stopped on must be reachable on the ladder; \
a cursor sealed past it strands it on every retry"
);
}
/// A ceiling reached on the FINAL page must still mint a continuation.
///
/// `pager.exhausted` breaks at the top of the loop AHEAD of every mint arm, so a
/// probe or visit ceiling that taints inside `gate_and_serve` while the last page is
/// being walked sheds `search_incomplete` with no token at all. `gl ipfs get` reads a
/// tokenless shed as "the ladder is over" (that is the wrapped-scan contract), so a
/// holder on that page is unreachable, permanently, on an inventory that never
/// changes.
#[cfg(unix)]
#[sqlx::test]
async fn get_by_cid_probe_ceiling_on_the_final_page_still_mints_a_continuation(
pool: sqlx::PgPool,
) {
let tmp = tempfile::TempDir::new().unwrap();
let mut state = crate::test_support::test_state(pool.clone()).await;
let repos_dir = tmp.path().join("repos");
std::fs::create_dir_all(&repos_dir).unwrap();
state.repo_store = crate::git::repo_store::RepoStore::for_testing(repos_dir, pool.clone());
state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
// One page holds the whole inventory, so the scan is exhausted the moment it
// starts and the break at `pager.exhausted` is the one that fires.
state.ipfs_legacy_scan_page_rows = 8;
state.ipfs_max_legacy_scan_rows = 1024;
state.ipfs_max_legacy_probes = 1;
state.ipfs_work_rate_limiter =
crate::rate_limit::RateLimiter::new(1024, std::time::Duration::from_secs(3600));
seed_root_readable_repos(&state, "finalpage", 1).await;
stamp_scan_order(&pool, "z6readablefinalpage/finalpage-0000", 0).await;
let (holder_id, oid) = seed_repo_with_blob(
&state,
tmp.path(),
"z6finalpage",
"holder",
b"on the last page\n",
)
.await;
stamp_scan_order(&pool, &holder_id, 1).await;
let cid = seed_legacy_pin_for_oid(&state, &oid).await;
let router = ipfs_router(state);
let peer: SocketAddr = "203.0.113.162:5000".parse().unwrap();
let (status, body) = status_and_body(
router
.clone()
.oneshot(get_cid_scan(&cid, Some(peer), None))
.await
.unwrap(),
)
.await;
assert_eq!(
status,
StatusCode::SERVICE_UNAVAILABLE,
"a probe ceiling on the final page is an incomplete search, not a verdict: {body}"
);
let token = continuation_of(&body).unwrap_or_else(|| {
panic!(
"a ceiling reached on the final page must still carry a continuation; \
a tokenless shed tells the caller their ladder is over: {body}"
)
});
let (status, body) = status_and_body(
router
.oneshot(get_cid_scan(&cid, Some(peer), Some(&token)))
.await
.unwrap(),
)
.await;
assert_eq!(
status,
StatusCode::OK,
"echoing the final-page continuation must reach the holder: {body}"
);
}
/// A ceiling on the final page of a RESUMED scan must keep its continuation.
///
/// `pager.resumed && pager.exhausted` is the wrapped-scan tail: the caller has walked
/// to the end of the table, so there is nothing left to resume and the absent token
/// is the signal. That is only true when the walk actually reached the end. A ceiling
/// stopping part way through the last page leaves rows unwalked in front of the
/// cursor, and clearing the seal there strands them exactly as a tokenless shed does.
///
/// Four rows, three per page, one probe: the third rung is the one that resumes into
/// a short page and stops on the holder.
///
/// MUTATION (RED): drop `scan_continuation.is_none()` from the wrap clause.
#[cfg(unix)]
#[sqlx::test]
async fn get_by_cid_ceiling_on_a_resumed_final_page_keeps_its_continuation(pool: sqlx::PgPool) {
let tmp = tempfile::TempDir::new().unwrap();
let mut state = crate::test_support::test_state(pool.clone()).await;
let repos_dir = tmp.path().join("repos");
std::fs::create_dir_all(&repos_dir).unwrap();
state.repo_store = crate::git::repo_store::RepoStore::for_testing(repos_dir, pool.clone());
state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
state.ipfs_legacy_scan_page_rows = 3;
state.ipfs_max_legacy_scan_rows = 1024;
state.ipfs_max_legacy_probes = 1;
state.ipfs_work_rate_limiter =
crate::rate_limit::RateLimiter::new(1024, std::time::Duration::from_secs(3600));
seed_root_readable_repos(&state, "wrapguard", 3).await;
for i in 0..3 {
stamp_scan_order(&pool, &format!("z6readablewrapguard/wrapguard-{i:04}"), i).await;
}
let (holder_id, oid) = seed_repo_with_blob(
&state,
tmp.path(),
"z6wrapguard",
"holder",
b"behind a resumed ceiling\n",
)
.await;
stamp_scan_order(&pool, &holder_id, 3).await;
let cid = seed_legacy_pin_for_oid(&state, &oid).await;
let router = ipfs_router(state);
let peer: SocketAddr = "203.0.113.163:5000".parse().unwrap();
let mut token: Option<String> = None;
let mut served_at = None;
for step in 1..=6 {
let (status, body) = status_and_body(
router
.clone()
.oneshot(get_cid_scan(&cid, Some(peer), token.as_deref()))
.await
.unwrap(),
)
.await;
if status == StatusCode::OK {
served_at = Some(step);
break;
}
assert_eq!(
status,
StatusCode::SERVICE_UNAVAILABLE,
"an intermediate rung is the retryable 503 (step {step}): {body}"
);
token = Some(
continuation_of(&body)
.unwrap_or_else(|| panic!("rung {step} must carry a continuation: {body}")),
);
}
assert!(
served_at.is_some(),
"a ceiling that stops part way through the last page of a resumed scan must \
still ladder; the wrap tail is for a walk that reached the end"
);
}
/// The VISIT ceiling must advance the ladder too, for the same reason as the probe
/// ceiling: it is the sibling arm, it fires on the same root-readable inventory, and
/// a tokenless shed there strands everything behind it just as permanently.