mirror of
https://github.com/Twigpine/node.git
synced 2026-11-08 15:39:37 -10:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c805e35f40 |
No files matched your search
+1
-1
@@ -81,7 +81,7 @@ time once updated.
|
||||
|
||||
**Requirements:**
|
||||
- Rust stable (≥ 1.91) — install via [rustup](https://rustup.rs)
|
||||
- PostgreSQL — required for the node. Use the bundled `docker-compose.yml` for local dev.
|
||||
- PostgreSQL 16+ — required for the node (it uses `pg_input_is_valid`, which exists only on PostgreSQL 16 and later). Use the bundled `docker-compose.yml` for local dev.
|
||||
- Docker (optional, for full-stack local testing)
|
||||
|
||||
**Environment variables:**
|
||||
|
||||
@@ -53,6 +53,7 @@ Good today:
|
||||
|
||||
- Local or Docker node startup.
|
||||
- Postgres-backed repo metadata.
|
||||
- Bounded GraphQL queries with [repository pagination](docs/graphql-pagination.md).
|
||||
- Bare git repository storage.
|
||||
- Git smart-HTTP clone/fetch/push.
|
||||
- RFC 9421-signed writes.
|
||||
|
||||
@@ -11,7 +11,8 @@ use crate::state::AppState;
|
||||
|
||||
/// Hard ceiling on rows any ref-update feed returns in one request. Shared by the
|
||||
/// shared collector's clamp and the per-handler request caps so they can't drift.
|
||||
const MAX_VISIBLE_REF_UPDATES: i64 = 200;
|
||||
/// `pub(crate)` so the GraphQL complexity meter can price against the same bound.
|
||||
pub(crate) const MAX_VISIBLE_REF_UPDATES: i64 = 200;
|
||||
|
||||
/// Collect up to `limit` ref-update rows visible to `caller`, newest first,
|
||||
/// paging past rows the feed gate drops. Filtering after a plain SQL `LIMIT`
|
||||
|
||||
@@ -8367,6 +8367,7 @@ mod tests {
|
||||
#[tokio::test]
|
||||
async fn get_by_cid_per_source_cap_sheds_same_source_admits_other() {
|
||||
let mut state = crate::test_support::test_state_lazy();
|
||||
state.db.pool().close().await;
|
||||
// Global pool has room; the per-source cap is 1.
|
||||
state.git_ipfs_walk_semaphore = Arc::new(Semaphore::new(8));
|
||||
state.git_ipfs_walk_per_caller = crate::rate_limit::PerCallerConcurrency::new(1, 100);
|
||||
@@ -8393,16 +8394,23 @@ mod tests {
|
||||
"a source at its per-source /ipfs walk cap must shed 503 with global capacity free"
|
||||
);
|
||||
|
||||
let bytes = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
|
||||
let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
|
||||
assert_eq!(body["error"], "overloaded");
|
||||
|
||||
// A DIFFERENT source is NOT shed by the per-source cap: it clears admission and
|
||||
// proceeds (then errors on the lazy DB, which is not a 503).
|
||||
// proceeds to the closed DB, which has a distinct db_unavailable error code.
|
||||
let resp = ipfs_router(state)
|
||||
.oneshot(get_cid(&cid, Some(other)))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_ne!(
|
||||
resp.status(),
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"a different source must not be shed by the per-source cap"
|
||||
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
|
||||
let bytes = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
|
||||
let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
|
||||
assert_eq!(
|
||||
body["error"],
|
||||
crate::error::DB_UNAVAILABLE_CODE,
|
||||
"a different source must clear admission and reach the closed database"
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -247,6 +247,40 @@ pub struct InfoRefsQuery {
|
||||
|
||||
// ── Handlers ──────────────────────────────────────────────────────────────
|
||||
|
||||
/// Validate a repository name at the create/fork write paths. The charset is
|
||||
/// restricted, and the length is bounded so every cursor the server emits
|
||||
/// parses back: `repo_cursor` embeds the name in a base64(JSON) cursor that
|
||||
/// `parse_repo_cursor` rejects over 4096 bytes (#465 4th round).
|
||||
///
|
||||
/// This is the API-side validator, and it deliberately differs from the
|
||||
/// stricter disk-side `validate_repo_name` in `git/repo_store.rs`: this one
|
||||
/// admits names the store rejects (empty, leading `-`, non-ASCII
|
||||
/// alphanumeric) and rejects `.`, which the store allows. The divergence
|
||||
/// predates this bound (#412 owns the fork half), so a store-rejected name
|
||||
/// still fails later at `repo_store.init` after the proof is spent. Do not
|
||||
/// "fix" the divergence here; the length bound below is the one that keeps
|
||||
/// emitted cursors parseable, and it is shared via `MAX_REPO_NAME_LEN` so the
|
||||
/// two validators cannot drift apart on length.
|
||||
fn validate_repo_name(name: &str) -> Result<()> {
|
||||
// Sanitize name: alphanumeric, hyphens, underscores only
|
||||
if !name
|
||||
.chars()
|
||||
.all(|c| c.is_alphanumeric() || c == '-' || c == '_')
|
||||
{
|
||||
return Err(AppError::BadRequest(
|
||||
"repo name must contain only alphanumeric characters, hyphens, and underscores".into(),
|
||||
));
|
||||
}
|
||||
// Byte length, not char count: the cursor budget is measured in bytes.
|
||||
if name.len() > crate::db::MAX_REPO_NAME_LEN {
|
||||
return Err(AppError::BadRequest(format!(
|
||||
"repo name must be at most {} bytes",
|
||||
crate::db::MAX_REPO_NAME_LEN
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// POST /api/v1/repos
|
||||
/// Create a new repository. Requires HTTP Signature auth.
|
||||
pub async fn create_repo(
|
||||
@@ -261,16 +295,7 @@ pub async fn create_repo(
|
||||
// rejected request (bad name, already exists) never burns a valid proof.
|
||||
let proof = crate::icaptcha::verify_request(&headers, &auth.0)?;
|
||||
|
||||
// Sanitize name: alphanumeric, hyphens, underscores only
|
||||
if !req
|
||||
.name
|
||||
.chars()
|
||||
.all(|c| c.is_alphanumeric() || c == '-' || c == '_')
|
||||
{
|
||||
return Err(AppError::BadRequest(
|
||||
"repo name must contain only alphanumeric characters, hyphens, and underscores".into(),
|
||||
));
|
||||
}
|
||||
validate_repo_name(&req.name)?;
|
||||
|
||||
// default_branch is caller-supplied and becomes a PR's target_branch when the
|
||||
// PR omits one, which is interpolated into a git revision argument. Validate
|
||||
@@ -3774,15 +3799,7 @@ pub async fn fork_repo(
|
||||
let fork_name = req.name.unwrap_or_else(|| source.name.clone());
|
||||
let forker_did = auth.0;
|
||||
|
||||
// Validate fork name
|
||||
if !fork_name
|
||||
.chars()
|
||||
.all(|c| c.is_alphanumeric() || c == '-' || c == '_')
|
||||
{
|
||||
return Err(AppError::BadRequest(
|
||||
"repo name must contain only alphanumeric characters, hyphens, and underscores".into(),
|
||||
));
|
||||
}
|
||||
validate_repo_name(&fork_name)?;
|
||||
|
||||
// Same barrier every other repo-creation route goes through. The character
|
||||
// allowlist above is vacuously true on an empty name, and `repo_disk_path`
|
||||
@@ -5049,6 +5066,50 @@ mod tests {
|
||||
peer_mock.assert_async().await;
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn repo_name_validation_rejects_bad_charset_and_overlong_names() {
|
||||
assert!(validate_repo_name("ok-name_123").is_ok());
|
||||
assert!(validate_repo_name(&"a".repeat(crate::db::MAX_REPO_NAME_LEN)).is_ok());
|
||||
// The bound is in bytes: the cursor budget is measured in bytes.
|
||||
let overlong = "a".repeat(crate::db::MAX_REPO_NAME_LEN + 1);
|
||||
let err = validate_repo_name(&overlong).unwrap_err();
|
||||
assert!(
|
||||
matches!(err, AppError::BadRequest(_)),
|
||||
"overlong name must be a 400, got: {err:?}"
|
||||
);
|
||||
assert!(validate_repo_name("bad name!").is_err());
|
||||
// Empty names were already admissible before the length bound; the
|
||||
// bound does not change that.
|
||||
assert!(validate_repo_name("").is_ok());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn create_repo_rejects_overlong_name_through_handler() {
|
||||
// Pins the wiring: `create_repo` must call `validate_repo_name`
|
||||
// before any database access. The lazy state never connects, so
|
||||
// deleting the call site would let the handler proceed to `get_repo`
|
||||
// and fail with a connection error instead of BadRequest.
|
||||
let state = crate::test_support::test_state_lazy();
|
||||
let overlong = "a".repeat(crate::db::MAX_REPO_NAME_LEN + 1);
|
||||
let err = create_repo(
|
||||
State(state),
|
||||
Extension(AuthenticatedDid(OWNER_DID.to_owned())),
|
||||
axum::http::HeaderMap::new(),
|
||||
Json(CreateRepoRequest {
|
||||
name: overlong,
|
||||
description: None,
|
||||
is_public: true,
|
||||
default_branch: "main".into(),
|
||||
}),
|
||||
)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
matches!(err, AppError::BadRequest(ref msg) if msg.contains("at most")),
|
||||
"overlong name through create_repo must be a 400, got: {err:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn upload_pack_request_finalizes_only_with_done_pktline() {
|
||||
let want = "0032want 1111111111111111111111111111111111111111\n";
|
||||
@@ -9179,6 +9240,7 @@ mod tests {
|
||||
/// A pkt-line receive-pack body carrying one branch-create ref update, so the
|
||||
/// handler's post-receive tail resolves a non-empty new-tip set (the delta
|
||||
/// scan's git stages run).
|
||||
#[cfg(unix)]
|
||||
fn ref_update_body(new_sha: &str) -> axum::body::Bytes {
|
||||
let line = format!("{ZERO_SHA} {new_sha} refs/heads/main");
|
||||
axum::body::Bytes::from(format!("{:04x}{}0000", line.len() + 4, line))
|
||||
@@ -11636,6 +11698,7 @@ mod tests {
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn f2a_log(log: &std::path::Path) -> String {
|
||||
std::fs::read_to_string(log).unwrap_or_default()
|
||||
}
|
||||
@@ -11643,6 +11706,7 @@ mod tests {
|
||||
/// Withheld-walk children run so far. `ls-tree` is the walk's signature child
|
||||
/// (`blob_paths` lists every reachable commit's tree); the delta scan and the
|
||||
/// full-scan fallback use `rev-list` / `cat-file` instead.
|
||||
#[cfg(unix)]
|
||||
fn f2a_walks(log: &std::path::Path) -> usize {
|
||||
f2a_log(log)
|
||||
.lines()
|
||||
@@ -11654,6 +11718,7 @@ mod tests {
|
||||
/// the withheld walk actually runs rather than taking the no-rule shortcut).
|
||||
/// The repo's on-disk path is passed to the tail directly, so no repo_store or
|
||||
/// receive-pack plumbing is involved.
|
||||
#[cfg(unix)]
|
||||
async fn f2a_state(
|
||||
pool: sqlx::PgPool,
|
||||
git_bin: &str,
|
||||
@@ -11685,6 +11750,7 @@ mod tests {
|
||||
(state, rec)
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn f2a_update(ref_name: &str, new_sha: &str) -> Vec<RefUpdate> {
|
||||
vec![RefUpdate {
|
||||
old_sha: ZERO_SHA.to_string(),
|
||||
@@ -11693,6 +11759,7 @@ mod tests {
|
||||
}]
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
const F2A_PUSHER: &str = "did:key:z6MkF2aPusherAAAAAAAAAAAAAAAAAAAAAAAAAA";
|
||||
|
||||
/// When publish durability is fenced, the tail must not run walks or take
|
||||
@@ -11805,6 +11872,7 @@ mod tests {
|
||||
}
|
||||
/// Poll `cond` until it holds, with a bound so a regression fails the test
|
||||
/// rather than hanging the suite.
|
||||
#[cfg(unix)]
|
||||
async fn f2a_wait_for(mut cond: impl FnMut() -> bool, what: &str) {
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
|
||||
while !cond() {
|
||||
@@ -11819,6 +11887,7 @@ mod tests {
|
||||
/// A `rev-list --objects` line names the tips a DELTA scan was asked to resolve,
|
||||
/// so it attributes that scan to one push's tips. The withheld walk's own
|
||||
/// `rev-list --all` / `ls-tree` lines never carry a tip as an argument this way.
|
||||
#[cfg(unix)]
|
||||
fn f2a_delta_scanned(log: &std::path::Path, tip: &str) -> bool {
|
||||
f2a_log(log)
|
||||
.lines()
|
||||
@@ -11967,6 +12036,7 @@ mod tests {
|
||||
|
||||
/// Mount a Pinata upload endpoint that assigns every object the same CID, and
|
||||
/// point the state at it. Returns the server (kept alive by the caller) and CID.
|
||||
#[cfg(unix)]
|
||||
async fn f2a_pinata(state: &mut AppState) -> (mockito::ServerGuard, String) {
|
||||
let cid = "bafyf2acoalescedmapping".to_string();
|
||||
let mut server = mockito::Server::new_async().await;
|
||||
@@ -11986,6 +12056,7 @@ mod tests {
|
||||
|
||||
/// Poll the branch to CID table until the push's mapping lands (the Pinata
|
||||
/// worker is detached), bounded so a regression fails rather than hangs.
|
||||
#[cfg(unix)]
|
||||
async fn f2a_wait_for_branch_cid(
|
||||
db: &crate::db::Db,
|
||||
slug: &str,
|
||||
@@ -12005,6 +12076,7 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn f2a_slug(rec: &crate::db::RepoRecord) -> String {
|
||||
format!(
|
||||
"{}/{}",
|
||||
@@ -12139,6 +12211,7 @@ mod tests {
|
||||
// and `z6p2fail`), and owner-only push is on by default, so the identity has to
|
||||
// follow the repo each push targets rather than being fixed for both.
|
||||
|
||||
#[cfg(unix)]
|
||||
fn p2_push(
|
||||
state: &AppState,
|
||||
owner: &str,
|
||||
@@ -12157,6 +12230,7 @@ mod tests {
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn p2_logged(log: &std::path::Path, prefix: &str) -> bool {
|
||||
f2a_log(log).lines().any(|l| l.starts_with(prefix))
|
||||
}
|
||||
@@ -13152,6 +13226,7 @@ exit 0
|
||||
/// runs exactly one `rev-list --all`, so this counts the walks that were attempted
|
||||
/// (the `ls-tree` counter above cannot: a walk whose enumeration fails never gets
|
||||
/// to `ls-tree`).
|
||||
#[cfg(unix)]
|
||||
fn f2b_walk_attempts(log: &std::path::Path) -> usize {
|
||||
f2a_log(log)
|
||||
.lines()
|
||||
|
||||
@@ -6,6 +6,20 @@ use std::time::Duration;
|
||||
use tracing::info;
|
||||
use uuid::Uuid;
|
||||
|
||||
/// Maximum visible repositories per page, plus one internal look-ahead row.
|
||||
pub(crate) const MAX_VISIBLE_REPO_PAGE_SIZE: usize = 200;
|
||||
|
||||
/// Maximum repository name length in bytes. `repo_cursor` embeds the name in
|
||||
/// a base64(JSON) cursor that `parse_repo_cursor` rejects over 4096 bytes, so
|
||||
/// every write path must bound the name: a cursor the server emits must
|
||||
/// always parse back (#465 4th round). 100 bytes keeps the longest possible
|
||||
/// cursor an order of magnitude under the parse limit.
|
||||
pub(crate) const MAX_REPO_NAME_LEN: usize = 100;
|
||||
|
||||
/// Maximum agent tasks returned per query. Shared by the GraphQL `tasks`
|
||||
/// complexity meter and the resolver clamp so they can't drift.
|
||||
pub(crate) const MAX_VISIBLE_TASK_PAGE_SIZE: i64 = 200;
|
||||
|
||||
// ── Public data types ─────────────────────────────────────────────────────────
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
@@ -405,6 +419,17 @@ impl Db {
|
||||
Self { pool }
|
||||
}
|
||||
|
||||
/// Test-only: access the underlying Postgres connection pool.
|
||||
///
|
||||
/// The production accessor was removed on main (ed76adf) once every
|
||||
/// production caller had migrated to typed `Db` methods; this restores
|
||||
/// it for tests only, for the raw-SQL fixtures and closed-pool cases in
|
||||
/// this change that have no typed equivalent.
|
||||
#[cfg(test)]
|
||||
pub fn pool(&self) -> &PgPool {
|
||||
&self.pool
|
||||
}
|
||||
|
||||
/// Test-only: apply the full schema to a fresh pool. `#[sqlx::test]`
|
||||
/// provisions an empty per-test database, so DB-backed tests must run this
|
||||
/// before seeding. Reuses the production `migrate()` path (the advisory lock
|
||||
@@ -1594,6 +1619,16 @@ impl Db {
|
||||
machine_id: Option<&str>,
|
||||
quarantined: bool,
|
||||
) -> Result<()> {
|
||||
// Peer-supplied names are bound like the local write paths: an
|
||||
// unbounded name would make reposPage emit a cursor longer than the
|
||||
// 4096-byte parse limit, dead-ending pagination for every caller past
|
||||
// that row (#465 4th round). The sync caller drops the error and skips
|
||||
// the mirror.
|
||||
if name.len() > MAX_REPO_NAME_LEN {
|
||||
anyhow::bail!(
|
||||
"mirror repo name exceeds {MAX_REPO_NAME_LEN} bytes: refusing to register"
|
||||
);
|
||||
}
|
||||
let now = Utc::now().to_rfc3339();
|
||||
let id = format!("{owner_short}/{name}");
|
||||
// `quarantined` is set only on first insert (the admission decision).
|
||||
@@ -1885,6 +1920,73 @@ impl Db {
|
||||
Ok(rows.into_iter().map(row_to_repo).collect())
|
||||
}
|
||||
|
||||
/// A bounded, mirror-deduplicated page ordered by owner key and repository
|
||||
/// name. Apply root visibility before LIMIT: private rows must not consume
|
||||
/// page slots or influence continuation metadata. The root-rule predicate
|
||||
/// mirrors `visibility::listable_at_root`; a differential test pins it.
|
||||
/// Malformed reader JSON denies that repo to non-owners without failing the page.
|
||||
/// Cursors only select a position and never confer read authority.
|
||||
///
|
||||
/// Performance note: the page orders and keyset-filters on
|
||||
/// `(owner_key, name)` under `COLLATE "C"`, but the existing
|
||||
/// `idx_repos_owner_key_name` index is built on the database default
|
||||
/// collation and cannot serve that ordering, so each page sorts the full
|
||||
/// deduped repo set before the keyset filter applies: O(total repos) work
|
||||
/// per page. This is accepted (pages are capped at
|
||||
/// `MAX_VISIBLE_REPO_PAGE_SIZE` rows over a narrow projection). A
|
||||
/// C-collation expression index was deliberately not added in a migration:
|
||||
/// versions 27-35 are claimed by other in-flight branches (Gitlawb/node#384
|
||||
/// claims 27-35, #464 claims 27-28) and the runner keys applied migrations
|
||||
/// on the version integer alone, so a colliding entry would be silently
|
||||
/// skipped on whichever side merges second. Revisit once that range lands
|
||||
/// or clears.
|
||||
pub async fn list_visible_repos_page(
|
||||
&self,
|
||||
caller: Option<&str>,
|
||||
after: Option<(&str, &str)>,
|
||||
limit: usize,
|
||||
) -> Result<Vec<RepoRecord>> {
|
||||
let sql = format!(
|
||||
"{}
|
||||
SELECT d.id, d.name, d.owner_did, d.description, d.is_public,
|
||||
d.default_branch, d.created_at, d.updated_at, d.disk_path,
|
||||
d.forked_from, d.machine_id
|
||||
FROM deduped d
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT reader_dids FROM visibility_rules v
|
||||
WHERE v.repo_id = d.id AND v.path_glob ~ '^/*([*][*])*$'
|
||||
ORDER BY v.path_glob DESC LIMIT 1
|
||||
) root_rule ON TRUE
|
||||
WHERE (
|
||||
($2::text IS NOT NULL AND ({key}) = $2)
|
||||
OR CASE WHEN root_rule.reader_dids IS NULL THEN d.is_public
|
||||
WHEN NOT pg_input_is_valid(root_rule.reader_dids, 'jsonb') THEN FALSE
|
||||
WHEN jsonb_typeof(root_rule.reader_dids::jsonb) = 'array' THEN
|
||||
COALESCE(root_rule.reader_dids::jsonb ? $3::text, FALSE)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM jsonb_array_elements(root_rule.reader_dids::jsonb) reader
|
||||
WHERE jsonb_typeof(reader) <> 'string'
|
||||
)
|
||||
ELSE FALSE END
|
||||
)
|
||||
AND ($4::text IS NULL OR (({key}) COLLATE \"C\", d.name COLLATE \"C\") > ($4, $5::text))
|
||||
ORDER BY ({key}) COLLATE \"C\", d.name COLLATE \"C\"
|
||||
LIMIT $6",
|
||||
Self::dedup_cte(),
|
||||
key = OWNER_KEY_CASE_SQL,
|
||||
);
|
||||
let rows = sqlx::query(&sql)
|
||||
.bind(None::<&str>)
|
||||
.bind(caller.map(normalize_owner_key))
|
||||
.bind(caller)
|
||||
.bind(after.map(|(owner, _)| normalize_owner_key(owner)))
|
||||
.bind(after.map(|(_, name)| name))
|
||||
.bind(limit.min(MAX_VISIBLE_REPO_PAGE_SIZE + 1) as i64)
|
||||
.fetch_all(&self.pool)
|
||||
.await?;
|
||||
Ok(rows.into_iter().map(row_to_repo).collect())
|
||||
}
|
||||
|
||||
/// Repos currently quarantined (admitted as mirrors but withheld from every
|
||||
/// listing surface). `list_all_repos_deduped` excludes these (its `DEDUP_CTE`
|
||||
/// filters `quarantined = FALSE`), so a gate that resolves a slug against the
|
||||
|
||||
@@ -118,7 +118,7 @@ impl RepoStore {
|
||||
/// Test-only: every guard from this store parks in `release` right before the
|
||||
/// `pg_advisory_unlock` await, until `gate` is notified. Dropping the future
|
||||
/// while it is parked reproduces a client disconnect inside `release`.
|
||||
#[cfg(test)]
|
||||
#[cfg(all(test, unix))]
|
||||
pub fn with_pre_unlock_gate(mut self, gate: Arc<tokio::sync::Notify>) -> Self {
|
||||
self.pre_unlock_gate = Some(gate);
|
||||
self
|
||||
@@ -126,7 +126,7 @@ impl RepoStore {
|
||||
|
||||
/// Test-only: how many write guards from this store have reached the Tigris upload
|
||||
/// site. See [`RepoWriteGuard::release`].
|
||||
#[cfg(test)]
|
||||
#[cfg(all(test, unix))]
|
||||
pub fn tigris_upload_site_reached(&self) -> usize {
|
||||
self.upload_site_reached
|
||||
.load(std::sync::atomic::Ordering::SeqCst)
|
||||
@@ -1717,8 +1717,11 @@ fn validate_repo_name(repo_name: &str) -> Result<()> {
|
||||
if repo_name.is_empty() {
|
||||
anyhow::bail!("repo_name is empty");
|
||||
}
|
||||
if repo_name.len() > 100 {
|
||||
anyhow::bail!("repo_name exceeds 100 chars");
|
||||
// Byte bound shared with the API-side validator via `MAX_REPO_NAME_LEN`:
|
||||
// the two validators differ on charset by design, but must not drift on
|
||||
// length, or an API-admitted name fails here after the proof is spent.
|
||||
if repo_name.len() > crate::db::MAX_REPO_NAME_LEN {
|
||||
anyhow::bail!("repo_name exceeds {} chars", crate::db::MAX_REPO_NAME_LEN);
|
||||
}
|
||||
// Repo names are `[A-Za-z0-9._-]+` minus path-traversal traps.
|
||||
if repo_name.contains("..") {
|
||||
|
||||
@@ -3,7 +3,7 @@ pub mod query;
|
||||
pub mod subscription;
|
||||
pub mod types;
|
||||
|
||||
use async_graphql::Schema;
|
||||
use async_graphql::{Schema, SchemaBuilder};
|
||||
use std::sync::Arc;
|
||||
|
||||
use crate::db::Db;
|
||||
@@ -14,6 +14,22 @@ use subscription::SubscriptionRoot;
|
||||
|
||||
pub type GitlawbSchema = Schema<QueryRoot, MutationRoot, SubscriptionRoot>;
|
||||
|
||||
// Keep ordinary schema discovery and application queries usable while
|
||||
// bounding validation work and the number of resolver selections per request.
|
||||
const GRAPHQL_MAX_COMPLEXITY: usize = 400;
|
||||
// The current public schema is shallow; this leaves headroom for composed
|
||||
// clients and the canonical getIntrospectionQuery without allowing
|
||||
// recursively nested documents to grow unchecked.
|
||||
const GRAPHQL_MAX_DEPTH: usize = 14;
|
||||
|
||||
fn apply_query_limits<Query, Mutation, Subscription>(
|
||||
builder: SchemaBuilder<Query, Mutation, Subscription>,
|
||||
) -> SchemaBuilder<Query, Mutation, Subscription> {
|
||||
builder
|
||||
.limit_complexity(GRAPHQL_MAX_COMPLEXITY)
|
||||
.limit_depth(GRAPHQL_MAX_DEPTH)
|
||||
}
|
||||
|
||||
/// Client-facing message for GraphQL resolver failures that wrap a real
|
||||
/// `sqlx::Error`. The real error is logged server-side; never put sqlx/Postgres
|
||||
/// detail in the GraphQL `errors` array (#250).
|
||||
@@ -81,7 +97,7 @@ pub fn build_schema(
|
||||
ref_update_tx: tokio::sync::broadcast::Sender<RefUpdateBroadcast>,
|
||||
task_event_tx: tokio::sync::broadcast::Sender<TaskEventBroadcast>,
|
||||
) -> GitlawbSchema {
|
||||
Schema::build(QueryRoot, MutationRoot, SubscriptionRoot)
|
||||
apply_query_limits(Schema::build(QueryRoot, MutationRoot, SubscriptionRoot))
|
||||
.data(db)
|
||||
.data(ref_update_tx)
|
||||
.data(task_event_tx)
|
||||
@@ -91,6 +107,10 @@ pub fn build_schema(
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::api::events::MAX_VISIBLE_REF_UPDATES;
|
||||
use crate::db::{MAX_VISIBLE_REPO_PAGE_SIZE, MAX_VISIBLE_TASK_PAGE_SIZE};
|
||||
use async_graphql::{EmptyMutation, EmptySubscription, Object, Value};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
#[test]
|
||||
fn graphql_db_err_opaques_sqlx_chain() {
|
||||
@@ -160,6 +180,668 @@ mod tests {
|
||||
assert!(!err.message.contains("failed to open"));
|
||||
}
|
||||
|
||||
fn production_test_schema() -> GitlawbSchema {
|
||||
let pool = sqlx::postgres::PgPoolOptions::new()
|
||||
.connect_lazy("postgres://localhost/unused")
|
||||
.unwrap();
|
||||
production_test_schema_with_db(Arc::new(Db::for_testing(pool)))
|
||||
}
|
||||
|
||||
fn production_test_schema_with_db(db: Arc<Db>) -> GitlawbSchema {
|
||||
build_schema(
|
||||
db,
|
||||
tokio::sync::broadcast::channel(1).0,
|
||||
tokio::sync::broadcast::channel(1).0,
|
||||
)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ref_update_subscription_alias_budget() {
|
||||
assert_subscription_alias_budget("refUpdates { repo }").await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn task_event_subscription_alias_budget() {
|
||||
assert_subscription_alias_budget("taskEvents { taskId }").await;
|
||||
}
|
||||
|
||||
async fn assert_subscription_alias_budget(field: &str) {
|
||||
use futures::StreamExt;
|
||||
use std::time::Duration;
|
||||
let pool = sqlx::postgres::PgPoolOptions::new()
|
||||
.connect_lazy("postgres://localhost/unused")
|
||||
.unwrap();
|
||||
let (ref_tx, _) = tokio::sync::broadcast::channel(16);
|
||||
let (task_tx, _) = tokio::sync::broadcast::channel(16);
|
||||
let schema = build_schema(
|
||||
Arc::new(Db::for_testing(pool)),
|
||||
ref_tx.clone(),
|
||||
task_tx.clone(),
|
||||
);
|
||||
for count in [7, 8] {
|
||||
let fields = (0..count)
|
||||
.map(|n| format!("r{n}: {field}"))
|
||||
.collect::<Vec<_>>()
|
||||
.join(" ");
|
||||
let query = format!("subscription {{ {fields} }}");
|
||||
let mut responses = Box::pin(schema.execute_stream(query));
|
||||
if count == 8 {
|
||||
let response = tokio::time::timeout(Duration::from_secs(1), responses.next())
|
||||
.await
|
||||
.expect("over-budget subscriptions must reject before waiting for events")
|
||||
.expect("the validation error must be returned");
|
||||
assert_eq!(response.errors.len(), 1);
|
||||
assert_eq!(response.errors[0].message, "Query is too complex.");
|
||||
assert_eq!(ref_tx.receiver_count() + task_tx.receiver_count(), 0);
|
||||
continue;
|
||||
}
|
||||
// Poll the accepted operation far enough to register its receivers.
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(30), responses.next())
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
assert_eq!(ref_tx.receiver_count() + task_tx.receiver_count(), 7);
|
||||
let _ = ref_tx.send(RefUpdateBroadcast {
|
||||
repo: "owner/repo".into(),
|
||||
ref_name: "refs/heads/main".into(),
|
||||
old_sha: "0".repeat(40),
|
||||
new_sha: "1".repeat(40),
|
||||
pusher_did: "did:key:test".into(),
|
||||
node_did: "did:key:node".into(),
|
||||
timestamp: "2026-01-01T00:00:00Z".into(),
|
||||
owner_did: "did:key:test".into(),
|
||||
});
|
||||
let _ = task_tx.send(TaskEventBroadcast {
|
||||
task_id: "task".into(),
|
||||
old_status: "pending".into(),
|
||||
new_status: "claimed".into(),
|
||||
by_did: "did:key:test".into(),
|
||||
at: "2026-01-01T00:00:00Z".into(),
|
||||
});
|
||||
let response = tokio::time::timeout(Duration::from_secs(1), responses.next())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(response.errors.is_empty(), "{:?}", response.errors);
|
||||
assert_ne!(response.data, Value::Null);
|
||||
drop(responses);
|
||||
assert_eq!(ref_tx.receiver_count() + task_tx.receiver_count(), 0);
|
||||
}
|
||||
}
|
||||
|
||||
#[sqlx::test]
|
||||
async fn production_mutation_budget_accepts_seven_and_rejects_eight(pool: sqlx::PgPool) {
|
||||
let db = Arc::new(Db::for_testing(pool));
|
||||
db.run_migrations().await.unwrap();
|
||||
let caller = "did:key:reviewer";
|
||||
let now = chrono::Utc::now().to_rfc3339();
|
||||
for index in 0..8 {
|
||||
db.create_task(&crate::db::AgentTask {
|
||||
id: format!("task-{index}"),
|
||||
repo_id: None,
|
||||
kind: "test".into(),
|
||||
status: "pending".into(),
|
||||
delegator_did: caller.into(),
|
||||
assignee_did: None,
|
||||
capability: "test".into(),
|
||||
ucan_token: None,
|
||||
payload: None,
|
||||
result: None,
|
||||
created_at: now.clone(),
|
||||
updated_at: now.clone(),
|
||||
deadline: None,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
let schema = production_test_schema_with_db(db.clone());
|
||||
for count in [8, 7] {
|
||||
let fields = (0..count)
|
||||
.map(|n| {
|
||||
format!("r{n}: claimTask(id: \"task-{n}\", assigneeDid: \"{caller}\") {{ id }}")
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
.join(" ");
|
||||
let response = schema
|
||||
.execute(
|
||||
async_graphql::Request::new(format!("mutation {{ {fields} }}"))
|
||||
.data(crate::auth::AuthenticatedDid(caller.into())),
|
||||
)
|
||||
.await;
|
||||
if count == 8 {
|
||||
assert_eq!(response.errors.len(), 1);
|
||||
assert_eq!(response.errors[0].message, "Query is too complex.");
|
||||
assert_eq!(
|
||||
db.list_tasks(Some("pending"), None, 8).await.unwrap().len(),
|
||||
8
|
||||
);
|
||||
continue;
|
||||
}
|
||||
assert!(response.errors.is_empty(), "{:?}", response.errors);
|
||||
assert_eq!(
|
||||
response
|
||||
.data
|
||||
.into_json()
|
||||
.unwrap()
|
||||
.as_object()
|
||||
.unwrap()
|
||||
.len(),
|
||||
7
|
||||
);
|
||||
assert_eq!(
|
||||
db.list_tasks(Some("claimed"), Some(caller), 8)
|
||||
.await
|
||||
.unwrap()
|
||||
.len(),
|
||||
7
|
||||
);
|
||||
assert_eq!(
|
||||
db.get_task("task-7").await.unwrap().unwrap().status,
|
||||
"pending"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[sqlx::test]
|
||||
async fn production_list_cost_scales_with_the_same_alias_count(pool: sqlx::PgPool) {
|
||||
let db = Arc::new(Db::for_testing(pool));
|
||||
db.run_migrations().await.unwrap();
|
||||
let schema = production_test_schema_with_db(db);
|
||||
// Bounds are per-root named constants so the cases move with the
|
||||
// documented maximums instead of going stale as literals.
|
||||
for (field, selection, max) in [
|
||||
("refUpdates", "repo", MAX_VISIBLE_REF_UPDATES),
|
||||
("tasks", "id", MAX_VISIBLE_TASK_PAGE_SIZE),
|
||||
(
|
||||
"reposPage",
|
||||
"nodes { name }",
|
||||
MAX_VISIBLE_REPO_PAGE_SIZE as i64,
|
||||
),
|
||||
] {
|
||||
// Two aliases at the maximum always exceed the budget.
|
||||
let fields = (0..2)
|
||||
.map(|n| format!("r{n}: {field}(limit: {max}) {{ {selection} }}"))
|
||||
.collect::<Vec<_>>()
|
||||
.join(" ");
|
||||
let response = schema.execute(format!("{{ {fields} }}")).await;
|
||||
assert_eq!(response.errors.len(), 1);
|
||||
assert_eq!(response.errors[0].message, "Query is too complex.");
|
||||
|
||||
// At limit 1 two aliases fit for the row-priced roots; reposPage
|
||||
// prices the full corpus scan, so a single alias is the most that
|
||||
// fits under the budget.
|
||||
let aliases = if field == "reposPage" { 1 } else { 2 };
|
||||
let fields = (0..aliases)
|
||||
.map(|n| format!("r{n}: {field}(limit: 1) {{ {selection} }}"))
|
||||
.collect::<Vec<_>>()
|
||||
.join(" ");
|
||||
let response = schema.execute(format!("{{ {fields} }}")).await;
|
||||
assert!(response.errors.is_empty(), "{field}: {:?}", response.errors);
|
||||
assert_eq!(
|
||||
response
|
||||
.data
|
||||
.into_json()
|
||||
.unwrap()
|
||||
.as_object()
|
||||
.unwrap()
|
||||
.len(),
|
||||
aliases
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn production_limits_reject_mutation_aliases_and_large_lists() {
|
||||
let schema = production_test_schema();
|
||||
// Limits are derived from the MAX_* constants so the boundaries move
|
||||
// with the documented maximums instead of going stale as literals.
|
||||
for (prefix, count, field) in [
|
||||
("mutation", 8, "claimTask(id: \"missing\", assigneeDid: \"did:key:test\") { id }".to_owned()),
|
||||
("mutation", 8, "createTask(delegatorDid: \"did:key:test\", input: { kind: \"test\", capability: \"test\" }) { id }".to_owned()),
|
||||
("mutation", 8, "completeTask(id: \"missing\", byDid: \"did:key:test\", input: {}) { id }".to_owned()),
|
||||
("mutation", 8, "failTask(id: \"missing\", byDid: \"did:key:test\", input: {}) { id }".to_owned()),
|
||||
("query", 8, "task(id: \"missing\") { id }".to_owned()),
|
||||
("query", 2, format!("refUpdates(limit: {MAX_VISIBLE_REF_UPDATES}) {{ repo }}")),
|
||||
("query", 2, format!("tasks(limit: {MAX_VISIBLE_TASK_PAGE_SIZE}) {{ id }}")),
|
||||
("query", 2, format!("reposPage(limit: {MAX_VISIBLE_REPO_PAGE_SIZE}) {{ nodes {{ name }} }}")),
|
||||
("query", 8, "reposPage(limit: 1) { nodes { name } }".to_owned()),
|
||||
] {
|
||||
let fields = (0..count)
|
||||
.map(|n| format!("r{n}: {field}"))
|
||||
.collect::<Vec<_>>()
|
||||
.join(" ");
|
||||
let response = schema.execute(format!("{prefix} {{ {fields} }}")).await;
|
||||
assert_eq!(response.errors.len(), 1, "{:?}", response.errors);
|
||||
assert_eq!(response.errors[0].message, "Query is too complex.");
|
||||
}
|
||||
}
|
||||
|
||||
#[sqlx::test]
|
||||
async fn production_repos_page_serves_the_documented_maximum(pool: sqlx::PgPool) {
|
||||
let db = Arc::new(Db::for_testing(pool));
|
||||
db.run_migrations().await.unwrap();
|
||||
let schema = production_test_schema_with_db(db);
|
||||
for selection in [
|
||||
"nodes { name }",
|
||||
"nodes { name ownerDid } hasNextPage endCursor",
|
||||
] {
|
||||
let response = schema
|
||||
.execute(format!(
|
||||
"{{ reposPage(limit: {MAX_VISIBLE_REPO_PAGE_SIZE}) {{ {selection} }} }}"
|
||||
))
|
||||
.await;
|
||||
assert!(response.errors.is_empty(), "{:?}", response.errors);
|
||||
assert_eq!(
|
||||
response.data.into_json().unwrap()["reposPage"]["nodes"],
|
||||
serde_json::json!([])
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn seven_root_aliases_are_accepted() {
|
||||
struct Root(Arc<AtomicUsize>);
|
||||
#[Object]
|
||||
impl Root {
|
||||
#[graphql(complexity = "50 + child_complexity")]
|
||||
async fn repos(&self) -> Vec<Nested> {
|
||||
self.0.fetch_add(1, Ordering::Relaxed);
|
||||
vec![Nested]
|
||||
}
|
||||
}
|
||||
let calls = Arc::new(AtomicUsize::new(0));
|
||||
let schema = apply_query_limits(Schema::build(
|
||||
Root(calls.clone()),
|
||||
EmptyMutation,
|
||||
EmptySubscription,
|
||||
))
|
||||
.finish();
|
||||
let fields = (0..7)
|
||||
.map(|n| format!("r{n}: repos {{ value }}"))
|
||||
.collect::<Vec<_>>()
|
||||
.join(" ");
|
||||
let response = schema.execute(format!("{{ {fields} }}")).await;
|
||||
assert!(response.errors.is_empty(), "{:?}", response.errors);
|
||||
assert_eq!(calls.load(Ordering::Relaxed), 7);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn expensive_root_aliases_are_rejected_before_database_access() {
|
||||
// Use the production builder with a lazy pool: rejection must precede DB access.
|
||||
let schema = production_test_schema();
|
||||
let fields = (0..8)
|
||||
.map(|n| format!("r{n}: repos {{ name }}"))
|
||||
.collect::<Vec<_>>()
|
||||
.join(" ");
|
||||
|
||||
let response = schema.execute(format!("{{ {fields} }}")).await;
|
||||
|
||||
assert_eq!(response.data, Value::Null);
|
||||
assert_eq!(response.errors.len(), 1);
|
||||
assert_eq!(response.errors[0].message, "Query is too complex.");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ordinary_schema_introspection_remains_available() {
|
||||
let schema = production_test_schema();
|
||||
let canonical_introspection = r#"
|
||||
query IntrospectionQuery {
|
||||
__schema {
|
||||
queryType { name }
|
||||
mutationType { name }
|
||||
subscriptionType { name }
|
||||
types {
|
||||
...FullType
|
||||
}
|
||||
directives {
|
||||
name
|
||||
description
|
||||
locations
|
||||
args {
|
||||
...InputValue
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fragment FullType on __Type {
|
||||
kind
|
||||
name
|
||||
description
|
||||
fields(includeDeprecated: true) {
|
||||
name
|
||||
description
|
||||
args {
|
||||
...InputValue
|
||||
}
|
||||
type {
|
||||
...TypeRef
|
||||
}
|
||||
isDeprecated
|
||||
deprecationReason
|
||||
}
|
||||
inputFields {
|
||||
...InputValue
|
||||
}
|
||||
interfaces {
|
||||
...TypeRef
|
||||
}
|
||||
enumValues(includeDeprecated: true) {
|
||||
name
|
||||
description
|
||||
isDeprecated
|
||||
deprecationReason
|
||||
}
|
||||
possibleTypes {
|
||||
...TypeRef
|
||||
}
|
||||
}
|
||||
|
||||
fragment InputValue on __InputValue {
|
||||
name
|
||||
description
|
||||
type { ...TypeRef }
|
||||
defaultValue
|
||||
}
|
||||
|
||||
fragment TypeRef on __Type {
|
||||
kind
|
||||
name
|
||||
ofType {
|
||||
kind
|
||||
name
|
||||
ofType {
|
||||
kind
|
||||
name
|
||||
ofType {
|
||||
kind
|
||||
name
|
||||
ofType {
|
||||
kind
|
||||
name
|
||||
ofType {
|
||||
kind
|
||||
name
|
||||
ofType {
|
||||
kind
|
||||
name
|
||||
ofType {
|
||||
kind
|
||||
name
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"#;
|
||||
let response = schema.execute(canonical_introspection).await;
|
||||
assert!(
|
||||
response.errors.is_empty(),
|
||||
"canonical introspection query failed against production schema: {:?}",
|
||||
response.errors
|
||||
);
|
||||
assert_ne!(response.data, Value::Null);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ref_updates_and_tasks_single_alias_max_limit_complexity_contract() {
|
||||
let pool = sqlx::postgres::PgPoolOptions::new()
|
||||
.connect_lazy("postgres://localhost/unused")
|
||||
.unwrap();
|
||||
pool.close().await;
|
||||
let (ref_tx, _) = tokio::sync::broadcast::channel(16);
|
||||
let (task_tx, _) = tokio::sync::broadcast::channel(16);
|
||||
let schema = build_schema(Arc::new(Db::for_testing(pool)), ref_tx, task_tx);
|
||||
|
||||
// refUpdates(limit: MAX_VISIBLE_REF_UPDATES):
|
||||
// Single-field selection has child_complexity 1 -> cost 100 + 200 + 1 = 301 <= 400.
|
||||
// It passes validation and reaches the resolver (which yields db error on lazy pool).
|
||||
let single_ref = schema
|
||||
.execute(format!(
|
||||
"{{ refUpdates(limit: {MAX_VISIBLE_REF_UPDATES}) {{ repo }} }}"
|
||||
))
|
||||
.await;
|
||||
assert!(
|
||||
!single_ref
|
||||
.errors
|
||||
.iter()
|
||||
.any(|e| e.message == "Query is too complex."),
|
||||
"single-field refUpdates at max limit must pass complexity validation: {:?}",
|
||||
single_ref.errors
|
||||
);
|
||||
assert!(
|
||||
single_ref
|
||||
.errors
|
||||
.iter()
|
||||
.any(|e| e.message == GRAPHQL_DB_ERROR_MESSAGE),
|
||||
"single-field refUpdates at max limit must reach the resolver: {:?}",
|
||||
single_ref.errors
|
||||
);
|
||||
|
||||
// Two-field selection has child_complexity 2 -> cost 100 + 200 + 2 = 302 <= 400.
|
||||
// The documented max-200 request stays reachable with multiple fields
|
||||
// (the old multiplicative price rejected it at 450); it passes
|
||||
// validation and reaches the resolver like the single-field case.
|
||||
let multi_ref = schema
|
||||
.execute(format!(
|
||||
"{{ refUpdates(limit: {MAX_VISIBLE_REF_UPDATES}) {{ repo refName }} }}"
|
||||
))
|
||||
.await;
|
||||
assert!(
|
||||
!multi_ref
|
||||
.errors
|
||||
.iter()
|
||||
.any(|e| e.message == "Query is too complex."),
|
||||
"two-field refUpdates at max limit must pass complexity validation: {:?}",
|
||||
multi_ref.errors
|
||||
);
|
||||
assert!(
|
||||
multi_ref
|
||||
.errors
|
||||
.iter()
|
||||
.any(|e| e.message == GRAPHQL_DB_ERROR_MESSAGE),
|
||||
"two-field refUpdates at max limit must reach the resolver: {:?}",
|
||||
multi_ref.errors
|
||||
);
|
||||
|
||||
// tasks(limit: MAX_VISIBLE_TASK_PAGE_SIZE):
|
||||
// Single-field selection has child_complexity 1, so the cost is
|
||||
// 50 + MAX_VISIBLE_TASK_PAGE_SIZE + 1 <= 400.
|
||||
// It passes validation and reaches the resolver.
|
||||
let single_task = schema
|
||||
.execute(format!(
|
||||
"{{ tasks(limit: {MAX_VISIBLE_TASK_PAGE_SIZE}) {{ id }} }}"
|
||||
))
|
||||
.await;
|
||||
assert!(
|
||||
!single_task
|
||||
.errors
|
||||
.iter()
|
||||
.any(|e| e.message == "Query is too complex."),
|
||||
"single-field tasks at the maximum page size must pass complexity validation: {:?}",
|
||||
single_task.errors
|
||||
);
|
||||
assert!(
|
||||
single_task
|
||||
.errors
|
||||
.iter()
|
||||
.any(|e| e.message == GRAPHQL_DB_ERROR_MESSAGE),
|
||||
"single-field tasks at the maximum page size must reach the resolver: {:?}",
|
||||
single_task.errors
|
||||
);
|
||||
|
||||
// Two-field selection has child_complexity 2, so the cost is
|
||||
// 50 + MAX_VISIBLE_TASK_PAGE_SIZE + 2 <= 400.
|
||||
// A request at the documented maximum stays reachable with multiple
|
||||
// fields (the old multiplicative price rejected it at 450); it passes
|
||||
// validation and reaches the resolver like the single-field case.
|
||||
let multi_task = schema
|
||||
.execute(format!(
|
||||
"{{ tasks(limit: {MAX_VISIBLE_TASK_PAGE_SIZE}) {{ id status }} }}"
|
||||
))
|
||||
.await;
|
||||
assert!(
|
||||
!multi_task
|
||||
.errors
|
||||
.iter()
|
||||
.any(|e| e.message == "Query is too complex."),
|
||||
"two-field tasks at the maximum page size must pass complexity validation: {:?}",
|
||||
multi_task.errors
|
||||
);
|
||||
assert!(
|
||||
multi_task
|
||||
.errors
|
||||
.iter()
|
||||
.any(|e| e.message == GRAPHQL_DB_ERROR_MESSAGE),
|
||||
"two-field tasks at the maximum page size must reach the resolver: {:?}",
|
||||
multi_task.errors
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn two_repos_aliases_exceed_the_budget() {
|
||||
// `repos` is priced as the full page it scans
|
||||
// (50 + MAX_VISIBLE_REPO_PAGE_SIZE + child), so two `repos { name }`
|
||||
// aliases cost 2 * 251 = 502 > 400 and the document is rejected before
|
||||
// any resolver runs. If the price reverts to `50 + child_complexity`
|
||||
// the aliases cost 2 * 51 = 102, validation passes, and the failure
|
||||
// mode changes (resolver DB errors on the lazy pool): this test goes
|
||||
// red on revert.
|
||||
let schema = production_test_schema();
|
||||
let response = schema
|
||||
.execute("{ r0: repos { name } r1: repos { name } }")
|
||||
.await;
|
||||
assert_eq!(response.errors.len(), 1, "{:?}", response.errors);
|
||||
assert_eq!(response.errors[0].message, "Query is too complex.");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn seven_repos_page_aliases_exceed_the_budget() {
|
||||
// `reposPage` is priced as the full corpus scan
|
||||
// (50 + MAX_VISIBLE_REPO_PAGE_SIZE + child_complexity), so seven
|
||||
// `reposPage(limit: 1) { nodes { name } }` aliases cost
|
||||
// 7 * (50 + 200 + 2) = 1764 > 400 and the document is rejected
|
||||
// before any resolver runs. Under the previous row-priced formula
|
||||
// the same aliases cost 7 * (50 + 1 + 2) = 371, validation passes,
|
||||
// and the failure mode changes (resolver DB errors on the lazy
|
||||
// pool): this test goes red on revert. Eight aliases are rejected
|
||||
// under both formulas, so seven is the load-bearing boundary.
|
||||
let schema = production_test_schema();
|
||||
let fields = (0..7)
|
||||
.map(|n| format!("r{n}: reposPage(limit: 1) {{ nodes {{ name }} }}"))
|
||||
.collect::<Vec<_>>()
|
||||
.join(" ");
|
||||
let response = schema.execute(format!("{{ {fields} }}")).await;
|
||||
assert_eq!(response.errors.len(), 1, "{:?}", response.errors);
|
||||
assert_eq!(response.errors[0].message, "Query is too complex.");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ref_updates_fixed_base_is_charged() {
|
||||
// The 100 base in the refUpdates complexity formula charges for the
|
||||
// collector's fixed work (loading the deduped repo set, the
|
||||
// quarantined set, and the visibility rules, then walking up to 16
|
||||
// keyset pages) even at limit: 1. Four aliases at limit: 1 cost
|
||||
// 4 * (100 + 1 + 1) = 408 > 400 and must be rejected. If the base
|
||||
// disappears, the cost drops to 4 * (1 + 1) = 8 and this test fails.
|
||||
let schema = production_test_schema();
|
||||
let fields = (0..4)
|
||||
.map(|n| format!("r{n}: refUpdates(limit: 1) {{ repo }}"))
|
||||
.collect::<Vec<_>>()
|
||||
.join(" ");
|
||||
let response = schema.execute(format!("{{ {fields} }}")).await;
|
||||
assert_eq!(response.errors.len(), 1, "{:?}", response.errors);
|
||||
assert_eq!(response.errors[0].message, "Query is too complex.");
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
struct Nested;
|
||||
|
||||
#[Object]
|
||||
impl Nested {
|
||||
async fn child(&self) -> Nested {
|
||||
Nested
|
||||
}
|
||||
|
||||
async fn value(&self) -> i32 {
|
||||
1
|
||||
}
|
||||
}
|
||||
|
||||
struct CountingQuery(Arc<AtomicUsize>);
|
||||
|
||||
#[Object]
|
||||
impl CountingQuery {
|
||||
async fn nested(&self) -> Nested {
|
||||
self.0.fetch_add(1, Ordering::Relaxed);
|
||||
Nested
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn query_depth_limit_accepts_fourteen_and_rejects_fifteen() {
|
||||
let calls = Arc::new(AtomicUsize::new(0));
|
||||
let schema = apply_query_limits(Schema::build(
|
||||
CountingQuery(Arc::clone(&calls)),
|
||||
EmptyMutation,
|
||||
EmptySubscription,
|
||||
))
|
||||
.finish();
|
||||
let query_at_depth = |depth: usize| {
|
||||
let selection = (2..depth).fold("value".to_string(), |selection, _| {
|
||||
format!("child {{ {selection} }}")
|
||||
});
|
||||
format!("{{ nested {{ {selection} }} }}")
|
||||
};
|
||||
|
||||
let accepted = schema.execute(query_at_depth(GRAPHQL_MAX_DEPTH)).await;
|
||||
|
||||
assert!(
|
||||
accepted.errors.is_empty(),
|
||||
"depth {GRAPHQL_MAX_DEPTH} should be accepted: {:?}",
|
||||
accepted.errors
|
||||
);
|
||||
assert_eq!(calls.swap(0, Ordering::Relaxed), 1);
|
||||
|
||||
let rejected = schema.execute(query_at_depth(GRAPHQL_MAX_DEPTH + 1)).await;
|
||||
|
||||
assert_eq!(rejected.data, Value::Null);
|
||||
assert_eq!(rejected.errors.len(), 1);
|
||||
assert_eq!(rejected.errors[0].message, "Query is nested too deep.");
|
||||
assert_eq!(calls.load(Ordering::Relaxed), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn production_schema_enforces_depth_limit() {
|
||||
let schema = production_test_schema();
|
||||
// Depth 14 is accepted through the production build_schema path.
|
||||
let of_types_14 = (0..9).fold("name".to_string(), |acc, _| format!("ofType {{ {acc} }}"));
|
||||
let query_14 =
|
||||
format!("{{ __schema {{ types {{ fields {{ type {{ {of_types_14} }} }} }} }} }}");
|
||||
let accepted = schema.execute(&query_14).await;
|
||||
assert!(accepted.errors.is_empty(), "{:?}", accepted.errors);
|
||||
|
||||
// Construct an introspection query of depth 15 using __schema.
|
||||
// 1: __schema
|
||||
// 2: types
|
||||
// 3: fields
|
||||
// 4: type
|
||||
// 5..14: ofType (10 times)
|
||||
// 15: name
|
||||
// Total depth = 15 > GRAPHQL_MAX_DEPTH (14).
|
||||
let of_types = (0..10).fold("name".to_string(), |acc, _| format!("ofType {{ {acc} }}"));
|
||||
let query = format!("{{ __schema {{ types {{ fields {{ type {{ {of_types} }} }} }} }} }}");
|
||||
|
||||
let rejected = schema.execute(&query).await;
|
||||
assert_eq!(rejected.data, async_graphql::Value::Null);
|
||||
assert_eq!(rejected.errors.len(), 1);
|
||||
assert_eq!(rejected.errors[0].message, "Query is nested too deep.");
|
||||
}
|
||||
|
||||
/// Every `.map_err(` in the GraphQL query/mutation resolvers must route
|
||||
/// through the opaque helpers, or discard the error (`|_|`). Same source-
|
||||
/// scrape pattern as `api::authz_guard` (#255 review).
|
||||
|
||||
@@ -22,6 +22,7 @@ pub struct MutationRoot;
|
||||
|
||||
#[Object]
|
||||
impl MutationRoot {
|
||||
#[graphql(complexity = "50 + child_complexity")]
|
||||
async fn create_task(
|
||||
&self,
|
||||
ctx: &Context<'_>,
|
||||
@@ -89,6 +90,7 @@ impl MutationRoot {
|
||||
Ok(AgentTaskType::from(stored))
|
||||
}
|
||||
|
||||
#[graphql(complexity = "50 + child_complexity")]
|
||||
async fn claim_task(
|
||||
&self,
|
||||
ctx: &Context<'_>,
|
||||
@@ -126,6 +128,7 @@ impl MutationRoot {
|
||||
Ok(AgentTaskType::from(task))
|
||||
}
|
||||
|
||||
#[graphql(complexity = "50 + child_complexity")]
|
||||
async fn complete_task(
|
||||
&self,
|
||||
ctx: &Context<'_>,
|
||||
@@ -177,6 +180,7 @@ impl MutationRoot {
|
||||
Ok(AgentTaskType::from(task))
|
||||
}
|
||||
|
||||
#[graphql(complexity = "50 + child_complexity")]
|
||||
async fn fail_task(
|
||||
&self,
|
||||
ctx: &Context<'_>,
|
||||
|
||||
@@ -1,57 +1,154 @@
|
||||
use async_graphql::{Context, Object, Result};
|
||||
use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
|
||||
use std::sync::Arc;
|
||||
|
||||
use crate::db::Db;
|
||||
use crate::api::events::MAX_VISIBLE_REF_UPDATES;
|
||||
use crate::db::{Db, RepoRecord, MAX_VISIBLE_REPO_PAGE_SIZE, MAX_VISIBLE_TASK_PAGE_SIZE};
|
||||
|
||||
use super::types::{AgentTaskType, RefUpdateType, RepoType};
|
||||
use super::types::{AgentTaskType, RefUpdateType, RepoPageType, RepoType};
|
||||
|
||||
fn repo_type(repo: RepoRecord) -> RepoType {
|
||||
RepoType {
|
||||
name: repo.name,
|
||||
owner_did: repo.owner_did,
|
||||
description: repo.description,
|
||||
default_branch: repo.default_branch,
|
||||
created_at: repo.created_at.to_rfc3339(),
|
||||
}
|
||||
}
|
||||
|
||||
fn repo_cursor(repo: &RepoRecord) -> String {
|
||||
URL_SAFE_NO_PAD.encode(
|
||||
serde_json::to_vec(&(&repo.owner_did, &repo.name))
|
||||
.expect("a pair of strings is JSON serializable"),
|
||||
)
|
||||
}
|
||||
|
||||
fn parse_repo_cursor(cursor: &str) -> Result<(String, String)> {
|
||||
// A position contains only public response fields, not a database id or
|
||||
// authority. Every page re-evaluates visibility for the current caller.
|
||||
if cursor.len() > 4096 {
|
||||
return Err(async_graphql::Error::new("invalid repository cursor"));
|
||||
}
|
||||
let bytes = URL_SAFE_NO_PAD
|
||||
.decode(cursor)
|
||||
.map_err(|_| async_graphql::Error::new("invalid repository cursor"))?;
|
||||
let position: (String, String) = serde_json::from_slice(&bytes)
|
||||
.map_err(|_| async_graphql::Error::new("invalid repository cursor"))?;
|
||||
// PostgreSQL text cannot contain NUL. Reject it as malformed input before
|
||||
// binding either cursor component, rather than reporting a database error.
|
||||
if position.0.contains('\0') || position.1.contains('\0') {
|
||||
return Err(async_graphql::Error::new("invalid repository cursor"));
|
||||
}
|
||||
Ok(position)
|
||||
}
|
||||
|
||||
pub struct QueryRoot;
|
||||
|
||||
#[Object]
|
||||
impl QueryRoot {
|
||||
// DB-backed roots carry a base cost so aliases consume the request budget
|
||||
// even when each alias selects only one inexpensive response field.
|
||||
// `repos` performs the same full corpus scan as a max-size `reposPage`
|
||||
// (list_visible_repos_page with MAX_VISIBLE_REPO_PAGE_SIZE + 1), so it
|
||||
// is priced as the full page it is, not as a cheap alias.
|
||||
#[graphql(complexity = "50 + MAX_VISIBLE_REPO_PAGE_SIZE + child_complexity")]
|
||||
/// Complete visible repository list, up to the documented maximum number
|
||||
/// of entries. Larger lists must use reposPage; this field returns an
|
||||
/// error instead of truncating silently.
|
||||
async fn repos(&self, ctx: &Context<'_>) -> Result<Vec<RepoType>> {
|
||||
let db = ctx.data_unchecked::<Arc<Db>>();
|
||||
let repos = db
|
||||
.list_all_repos_deduped()
|
||||
.await
|
||||
.map_err(crate::graphql::graphql_db_err)?;
|
||||
|
||||
// Apply the same "/" visibility gate the REST/per-repo endpoints use so
|
||||
// this surface does not enumerate private repos (#97). The caller DID is
|
||||
// threaded onto the context by optional_signature; absent = anonymous.
|
||||
let caller = ctx
|
||||
.data::<crate::auth::AuthenticatedDid>()
|
||||
.ok()
|
||||
.map(|d| d.0.as_str());
|
||||
let ids: Vec<String> = repos.iter().map(|r| r.id.clone()).collect();
|
||||
let rules_by_repo = db
|
||||
.list_visibility_rules_for_repos(&ids)
|
||||
let mut repos = db
|
||||
.list_visible_repos_page(caller, None, MAX_VISIBLE_REPO_PAGE_SIZE + 1)
|
||||
.await
|
||||
.map_err(crate::graphql::graphql_db_err)?;
|
||||
|
||||
Ok(repos
|
||||
.into_iter()
|
||||
.filter(|r| {
|
||||
let rules = rules_by_repo.get(&r.id).map(Vec::as_slice).unwrap_or(&[]);
|
||||
crate::visibility::listable_at_root(rules, r.is_public, &r.owner_did, caller)
|
||||
})
|
||||
.map(|r| RepoType {
|
||||
name: r.name,
|
||||
owner_did: r.owner_did,
|
||||
description: r.description,
|
||||
default_branch: r.default_branch,
|
||||
created_at: r.created_at.to_rfc3339(),
|
||||
})
|
||||
.collect())
|
||||
if repos.len() > MAX_VISIBLE_REPO_PAGE_SIZE {
|
||||
return Err(async_graphql::Error::new(format!(
|
||||
"repository list exceeds {} entries; use reposPage with limit and after",
|
||||
MAX_VISIBLE_REPO_PAGE_SIZE
|
||||
)));
|
||||
}
|
||||
// Preserve the legacy activity ordering for complete, small lists.
|
||||
repos.sort_by_key(|repo| std::cmp::Reverse(repo.updated_at));
|
||||
Ok(repos.into_iter().map(repo_type).collect())
|
||||
}
|
||||
|
||||
/// Bounded visible repositories ordered by owner and name. Continue with
|
||||
/// endCursor while hasNextPage is true. Pages are not a database snapshot.
|
||||
// Every page sorts the full deduplicated corpus under COLLATE "C"
|
||||
// (list_visible_repos_page documents the accepted O(total repos) bound),
|
||||
// so the price charges the scan, not the emitted rows: a limit:1 page
|
||||
// costs the same as a max-size page, the way `repos` is priced.
|
||||
#[graphql(complexity = "50 + MAX_VISIBLE_REPO_PAGE_SIZE + child_complexity")]
|
||||
async fn repos_page(
|
||||
&self,
|
||||
ctx: &Context<'_>,
|
||||
#[graphql(
|
||||
default = 50,
|
||||
desc = "Page size from 1 to the documented maximum; other values are rejected."
|
||||
)]
|
||||
limit: i64,
|
||||
after: Option<String>,
|
||||
) -> Result<RepoPageType> {
|
||||
if !(1..=MAX_VISIBLE_REPO_PAGE_SIZE as i64).contains(&limit) {
|
||||
return Err(async_graphql::Error::new(format!(
|
||||
"limit must be between 1 and {}",
|
||||
MAX_VISIBLE_REPO_PAGE_SIZE
|
||||
)));
|
||||
}
|
||||
let after = after.as_deref().map(parse_repo_cursor).transpose()?;
|
||||
let caller = ctx
|
||||
.data::<crate::auth::AuthenticatedDid>()
|
||||
.ok()
|
||||
.map(|d| d.0.as_str());
|
||||
let db = ctx.data_unchecked::<Arc<Db>>();
|
||||
let mut repos = db
|
||||
.list_visible_repos_page(
|
||||
caller,
|
||||
after
|
||||
.as_ref()
|
||||
.map(|(owner, name)| (owner.as_str(), name.as_str())),
|
||||
limit as usize + 1,
|
||||
)
|
||||
.await
|
||||
.map_err(crate::graphql::graphql_db_err)?;
|
||||
let has_next_page = repos.len() > limit as usize;
|
||||
repos.truncate(limit as usize);
|
||||
let end_cursor = repos.last().map(repo_cursor);
|
||||
Ok(RepoPageType {
|
||||
nodes: repos.into_iter().map(repo_type).collect(),
|
||||
has_next_page,
|
||||
end_cursor,
|
||||
})
|
||||
}
|
||||
|
||||
// Complexity is additive with a base reflecting the collector's fixed scan:
|
||||
// every request loads the full deduped repo set, the quarantined set, and
|
||||
// the visibility rules, then walks up to max(limit, 2048) event rows past
|
||||
// withheld ones, regardless of `limit`. A multiplicative price both
|
||||
// undercharged limit:1 (51 for that fixed work) and made a request at the
|
||||
// documented maximum unreachable (two fields at max limit cost 450 > 400).
|
||||
//
|
||||
// Decision on the collector floor (review #465, 3rd round): the up-to-2048
|
||||
// walk is bounded constant overhead that does not scale with `limit`, so
|
||||
// it is priced in the fixed base, not the limit term; the limit term
|
||||
// prices only the emitted rows. Pricing the walk at face value in the base
|
||||
// would put every refUpdates query over the 400 budget and make the field
|
||||
// unusable, so the base prices that fixed scan as a lump sum instead.
|
||||
#[graphql(
|
||||
complexity = "100 + (limit.clamp(0, MAX_VISIBLE_REF_UPDATES) as usize) + child_complexity"
|
||||
)]
|
||||
async fn ref_updates(
|
||||
&self,
|
||||
ctx: &Context<'_>,
|
||||
repo: Option<String>,
|
||||
#[graphql(
|
||||
default = 20,
|
||||
desc = "Max 200; larger requests return the newest 200 rows (no continuation cursor)."
|
||||
desc = "Requests above the documented maximum return the newest rows up to that maximum (no continuation cursor)."
|
||||
)]
|
||||
limit: i64,
|
||||
) -> Result<Vec<RefUpdateType>> {
|
||||
@@ -105,6 +202,14 @@ impl QueryRoot {
|
||||
Ok(resolved)
|
||||
}
|
||||
|
||||
// Complexity is additive: list_tasks is a single LIMIT-bounded query whose
|
||||
// cost does not scale with the selected fields, so the price is the base
|
||||
// plus the clamped limit plus the child fields. A multiplicative price
|
||||
// made a request at the documented maximum unreachable (two fields at max
|
||||
// limit cost 450 > 400).
|
||||
#[graphql(
|
||||
complexity = "50 + (limit.clamp(0, MAX_VISIBLE_TASK_PAGE_SIZE) as usize) + child_complexity"
|
||||
)]
|
||||
async fn tasks(
|
||||
&self,
|
||||
ctx: &Context<'_>,
|
||||
@@ -112,7 +217,7 @@ impl QueryRoot {
|
||||
assignee_did: Option<String>,
|
||||
#[graphql(
|
||||
default = 50,
|
||||
desc = "Max 200; larger requests are clamped to 200 (no error). Negative values clamp to 0."
|
||||
desc = "Requests above the documented maximum are clamped to it (no error). Negative values clamp to 0."
|
||||
)]
|
||||
limit: i64,
|
||||
) -> Result<Vec<AgentTaskType>> {
|
||||
@@ -120,7 +225,7 @@ impl QueryRoot {
|
||||
// Clamp before SQL: a negative LIMIT is a client fault that Postgres
|
||||
// rejects with 2201W, which would otherwise trip the opaque DB path
|
||||
// and write an error-level log on every probe (#250 review).
|
||||
let limit = limit.clamp(0, 200);
|
||||
let limit = limit.clamp(0, MAX_VISIBLE_TASK_PAGE_SIZE);
|
||||
let tasks = db
|
||||
.list_tasks(status.as_deref(), assignee_did.as_deref(), limit)
|
||||
.await
|
||||
@@ -128,6 +233,7 @@ impl QueryRoot {
|
||||
Ok(tasks.into_iter().map(AgentTaskType::from).collect())
|
||||
}
|
||||
|
||||
#[graphql(complexity = "50 + child_complexity")]
|
||||
async fn task(&self, ctx: &Context<'_>, id: String) -> Result<Option<AgentTaskType>> {
|
||||
let db = ctx.data_unchecked::<Arc<Db>>();
|
||||
let t = db
|
||||
@@ -141,12 +247,642 @@ impl QueryRoot {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::db::{Db, ReceivedRefUpdate, RepoRecord};
|
||||
use base64::Engine;
|
||||
use chrono::Utc;
|
||||
use sqlx::PgPool;
|
||||
use std::sync::Arc;
|
||||
|
||||
const OWNER: &str = "did:key:z6MkOwner";
|
||||
|
||||
#[sqlx::test]
|
||||
async fn repos_legacy_rejects_overflow_and_pages_reach_every_visible_repo(pool: PgPool) {
|
||||
let db = db(pool).await;
|
||||
let total = crate::db::MAX_VISIBLE_REPO_PAGE_SIZE + 5;
|
||||
for index in 0..total {
|
||||
let name = format!("repo-{index:03}");
|
||||
db.create_repo(&repo(&name, OWNER, &name, true))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
// The SQL helper itself must bound materialization, even for a caller
|
||||
// accidentally requesting an unlimited page.
|
||||
assert_eq!(
|
||||
db.list_visible_repos_page(None, None, usize::MAX)
|
||||
.await
|
||||
.unwrap()
|
||||
.len(),
|
||||
crate::db::MAX_VISIBLE_REPO_PAGE_SIZE + 1
|
||||
);
|
||||
let schema = schema(db);
|
||||
let legacy = anon(&schema, "{ repos { name } }").await;
|
||||
assert_eq!(legacy.errors.len(), 1);
|
||||
assert!(legacy.errors[0].message.contains("use reposPage"));
|
||||
assert_eq!(legacy.data, async_graphql::Value::Null);
|
||||
|
||||
let mut cursor = None;
|
||||
let mut names = Vec::new();
|
||||
loop {
|
||||
let response = schema.execute(
|
||||
async_graphql::Request::new(
|
||||
"query($after: String) { reposPage(limit: 50, after: $after) { nodes { name } hasNextPage endCursor } }",
|
||||
).variables(async_graphql::Variables::from_json(serde_json::json!({"after": cursor}))),
|
||||
).await;
|
||||
assert!(response.errors.is_empty(), "{:?}", response.errors);
|
||||
let json = response.data.into_json().unwrap();
|
||||
let page = &json["reposPage"];
|
||||
let rows = page["nodes"].as_array().unwrap();
|
||||
assert!(rows.len() <= 50);
|
||||
names.extend(
|
||||
rows.iter()
|
||||
.map(|row| row["name"].as_str().unwrap().to_owned()),
|
||||
);
|
||||
if !page["hasNextPage"].as_bool().unwrap() {
|
||||
break;
|
||||
}
|
||||
let next = page["endCursor"].as_str().unwrap().to_owned();
|
||||
assert_ne!(cursor.as_ref(), Some(&next));
|
||||
cursor = Some(next);
|
||||
assert!(names.len() <= crate::db::MAX_VISIBLE_REPO_PAGE_SIZE);
|
||||
}
|
||||
assert_eq!(
|
||||
names,
|
||||
(0..total)
|
||||
.map(|index| format!("repo-{index:03}"))
|
||||
.collect::<Vec<_>>()
|
||||
);
|
||||
}
|
||||
|
||||
#[sqlx::test]
|
||||
async fn repos_legacy_accepts_exactly_the_visible_bound(pool: PgPool) {
|
||||
let db = db(pool).await;
|
||||
let base_time = Utc::now();
|
||||
let total = crate::db::MAX_VISIBLE_REPO_PAGE_SIZE;
|
||||
for index in 0..total {
|
||||
let name = format!("repo-{index:03}");
|
||||
let mut r = repo(&name, OWNER, &name, true);
|
||||
// Anti-correlate updated_at with name, creation order, and index
|
||||
// so only ordering by updated_at DESC can satisfy the expectation.
|
||||
let offset = (index * 37) % total;
|
||||
r.updated_at = base_time + chrono::Duration::seconds(offset as i64);
|
||||
db.create_repo(&r).await.unwrap();
|
||||
}
|
||||
db.create_repo(&repo("hidden", OWNER, "hidden", false))
|
||||
.await
|
||||
.unwrap();
|
||||
let response = anon(&schema(db), "{ repos { name } }").await;
|
||||
assert!(response.errors.is_empty(), "{:?}", response.errors);
|
||||
let repos = response.data.into_json().unwrap()["repos"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.clone();
|
||||
assert_eq!(repos.len(), total);
|
||||
let names: Vec<_> = repos.iter().filter_map(|r| r["name"].as_str()).collect();
|
||||
assert!(
|
||||
!names.contains(&"hidden"),
|
||||
"hidden repo must be excluded from legacy repos response"
|
||||
);
|
||||
let mut expected_indices = (0..total).collect::<Vec<_>>();
|
||||
expected_indices.sort_by_key(|&idx| std::cmp::Reverse((idx * 37) % total));
|
||||
let expected_names: Vec<String> = expected_indices
|
||||
.into_iter()
|
||||
.map(|index| format!("repo-{index:03}"))
|
||||
.collect();
|
||||
assert_eq!(
|
||||
names,
|
||||
expected_names
|
||||
.iter()
|
||||
.map(|s| s.as_str())
|
||||
.collect::<Vec<_>>(),
|
||||
"legacy repos must retain activity ordering (most recently updated first)"
|
||||
);
|
||||
}
|
||||
|
||||
#[sqlx::test]
|
||||
async fn repos_page_pagination_crosses_owner_boundary(pool: PgPool) {
|
||||
let db = db(pool).await;
|
||||
let owner1 = "did:key:z6MkaOwner1";
|
||||
let owner2 = "did:key:z6MkbOwner2";
|
||||
let earlier_owner = "did:web:example.com";
|
||||
// This owner sorts first only after did:key normalization. Its name
|
||||
// sorts past the first cursor, so an unqualified OR name predicate
|
||||
// would incorrectly serve it again on the next page.
|
||||
//
|
||||
// Rows are inserted out of order and not in reverse asserted order.
|
||||
// Assigned IDs do not track name order, and owner2 contains both
|
||||
// "B-repo" and "a-repo" whose byte order (B < a) differs from their
|
||||
// case-folded order (a < b), ensuring lower(d.name) cannot satisfy
|
||||
// the keyset query. Sized to 6 entries so the last page is full-sized
|
||||
// (limit 2), proving hasNextPage distinguishes a full terminal page.
|
||||
db.create_repo(&repo("r5", owner2, "c-repo", true))
|
||||
.await
|
||||
.unwrap();
|
||||
db.create_repo(&repo("r9", earlier_owner, "zz-repo", true))
|
||||
.await
|
||||
.unwrap();
|
||||
db.create_repo(&repo("r3", owner2, "B-repo", true))
|
||||
.await
|
||||
.unwrap();
|
||||
db.create_repo(&repo("r8", owner1, "z-repo", true))
|
||||
.await
|
||||
.unwrap();
|
||||
db.create_repo(&repo("r2", owner1, "b-repo", true))
|
||||
.await
|
||||
.unwrap();
|
||||
db.create_repo(&repo("r1", owner2, "a-repo", true))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let schema = schema(db);
|
||||
let page1_resp = anon(
|
||||
&schema,
|
||||
"{ reposPage(limit: 2) { nodes { name ownerDid } hasNextPage endCursor } }",
|
||||
)
|
||||
.await;
|
||||
assert!(page1_resp.errors.is_empty(), "{:?}", page1_resp.errors);
|
||||
let p1 = page1_resp.data.into_json().unwrap()["reposPage"].clone();
|
||||
assert_eq!(p1["hasNextPage"], true);
|
||||
assert_eq!(
|
||||
p1["nodes"],
|
||||
serde_json::json!([
|
||||
{"name": "zz-repo", "ownerDid": earlier_owner},
|
||||
{"name": "b-repo", "ownerDid": owner1},
|
||||
])
|
||||
);
|
||||
let cursor = p1["endCursor"].as_str().unwrap();
|
||||
|
||||
let page2_query = format!(
|
||||
"{{ reposPage(limit: 2, after: \"{cursor}\") {{ nodes {{ name ownerDid }} hasNextPage endCursor }} }}"
|
||||
);
|
||||
let page2_resp = anon(&schema, &page2_query).await;
|
||||
assert!(page2_resp.errors.is_empty(), "{:?}", page2_resp.errors);
|
||||
let p2 = page2_resp.data.into_json().unwrap()["reposPage"].clone();
|
||||
assert_eq!(p2["hasNextPage"], true);
|
||||
assert_eq!(
|
||||
p2["nodes"],
|
||||
serde_json::json!([
|
||||
{"name": "z-repo", "ownerDid": owner1},
|
||||
{"name": "B-repo", "ownerDid": owner2},
|
||||
])
|
||||
);
|
||||
let cursor = p2["endCursor"].as_str().unwrap();
|
||||
let page3_query = format!(
|
||||
"{{ reposPage(limit: 2, after: \"{cursor}\") {{ nodes {{ name ownerDid }} hasNextPage endCursor }} }}"
|
||||
);
|
||||
let page3_resp = anon(&schema, &page3_query).await;
|
||||
assert!(page3_resp.errors.is_empty(), "{:?}", page3_resp.errors);
|
||||
let p3 = page3_resp.data.into_json().unwrap()["reposPage"].clone();
|
||||
assert_eq!(p3["hasNextPage"], false);
|
||||
assert_eq!(
|
||||
p3["nodes"],
|
||||
serde_json::json!([
|
||||
{"name": "a-repo", "ownerDid": owner2},
|
||||
{"name": "c-repo", "ownerDid": owner2},
|
||||
])
|
||||
);
|
||||
}
|
||||
|
||||
#[sqlx::test]
|
||||
async fn repos_page_at_documented_maximum_distinguishes_terminal_page(pool: PgPool) {
|
||||
let db = db(pool).await;
|
||||
let total = crate::db::MAX_VISIBLE_REPO_PAGE_SIZE;
|
||||
for index in 0..total {
|
||||
let name = format!("repo-{index:03}");
|
||||
db.create_repo(&repo(&name, OWNER, &name, true))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
let schema = schema(db.clone());
|
||||
let query =
|
||||
format!("{{ reposPage(limit: {total}) {{ nodes {{ name }} hasNextPage endCursor }} }}");
|
||||
let resp200 = anon(&schema, &query).await;
|
||||
assert!(resp200.errors.is_empty(), "{:?}", resp200.errors);
|
||||
let p200 = resp200.data.into_json().unwrap()["reposPage"].clone();
|
||||
assert_eq!(p200["hasNextPage"], false);
|
||||
let nodes200 = p200["nodes"].as_array().unwrap().clone();
|
||||
assert_eq!(nodes200.len(), total);
|
||||
|
||||
// Adding row 201 keeps page 1 content identical while flipping hasNextPage to true.
|
||||
let name201 = format!("repo-{total:03}");
|
||||
db.create_repo(&repo(&name201, OWNER, &name201, true))
|
||||
.await
|
||||
.unwrap();
|
||||
let resp201 = anon(&schema, &query).await;
|
||||
assert!(resp201.errors.is_empty(), "{:?}", resp201.errors);
|
||||
let p201 = resp201.data.into_json().unwrap()["reposPage"].clone();
|
||||
assert_eq!(p201["hasNextPage"], true);
|
||||
assert_eq!(p201["nodes"], p200["nodes"]);
|
||||
assert_eq!(p201["endCursor"], p200["endCursor"]);
|
||||
}
|
||||
|
||||
#[sqlx::test]
|
||||
async fn repos_page_visibility_matches_the_shared_gate(pool: PgPool) {
|
||||
use crate::db::VisibilityMode;
|
||||
let db = db(pool).await;
|
||||
let reader = "did:key:zReader";
|
||||
for (id, public) in [
|
||||
("open", true),
|
||||
("private", false),
|
||||
("root-deny", true),
|
||||
("root-reader", false),
|
||||
("subtree", true),
|
||||
("root-tie", true),
|
||||
("odd-star", true),
|
||||
("malformed-readers", true),
|
||||
("non-array-readers", true),
|
||||
("non-string-reader", true),
|
||||
("quarantined", true),
|
||||
] {
|
||||
db.create_repo(&repo(id, OWNER, id, public)).await.unwrap();
|
||||
}
|
||||
// Canonical and mirror copies must still collapse before pagination.
|
||||
db.create_repo(&repo("z6MkOwner/open", "z6MkOwner", "open", true))
|
||||
.await
|
||||
.unwrap();
|
||||
db.create_repo(&repo(
|
||||
"other-method",
|
||||
"did:web:z6MkOwner",
|
||||
"other-method",
|
||||
false,
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
db.set_repo_quarantine("quarantined", true).await.unwrap();
|
||||
for (id, glob, readers) in [
|
||||
("root-deny", "/", vec![]),
|
||||
("root-reader", "/**", vec![reader.to_owned()]),
|
||||
("subtree", "/secret/**", vec![]),
|
||||
("root-tie", "/", vec![reader.to_owned()]),
|
||||
("root-tie", "/**", vec![]),
|
||||
("odd-star", "/*", vec![]),
|
||||
("malformed-readers", "/", vec![]),
|
||||
("non-array-readers", "/", vec![]),
|
||||
("non-string-reader", "/", vec![]),
|
||||
] {
|
||||
db.set_visibility_rule(id, glob, VisibilityMode::B, &readers, OWNER)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
// The typed setter cannot create malformed JSON, but existing TEXT rows
|
||||
// can contain it. Such a rule must deny this repo, not abort the page.
|
||||
// All three deny arms of the reader_dids predicate are pinned here:
|
||||
// (a) invalid JSON, (b) valid JSON that is not an array, (c) an array
|
||||
// containing a non-string member. The Rust gate parses reader_dids as
|
||||
// Vec<String> with unwrap_or_default(), so (b) and (c) also fail to
|
||||
// parse and deny; the SQL CASE must agree on every shape.
|
||||
let rows = sqlx::query("UPDATE visibility_rules SET reader_dids = $1 WHERE repo_id = $2")
|
||||
.bind("not JSON")
|
||||
.bind("malformed-readers")
|
||||
.execute(db.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(rows.rows_affected(), 1);
|
||||
let rows = sqlx::query("UPDATE visibility_rules SET reader_dids = $1 WHERE repo_id = $2")
|
||||
.bind("\"just-a-string\"")
|
||||
.bind("non-array-readers")
|
||||
.execute(db.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(rows.rows_affected(), 1);
|
||||
// The named reader is listed, but the non-string member poisons the
|
||||
// whole list: both sides must still deny them.
|
||||
let rows = sqlx::query("UPDATE visibility_rules SET reader_dids = $1 WHERE repo_id = $2")
|
||||
.bind("[\"did:key:zReader\", 42]")
|
||||
.bind("non-string-reader")
|
||||
.execute(db.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(rows.rows_affected(), 1);
|
||||
let all = db.list_all_repos_deduped().await.unwrap();
|
||||
for caller in [
|
||||
None,
|
||||
Some(OWNER),
|
||||
Some("z6MkOwner"),
|
||||
Some(reader),
|
||||
Some("zReader"),
|
||||
Some("did:web:z6MkOwner"),
|
||||
] {
|
||||
let mut expected = Vec::new();
|
||||
for record in &all {
|
||||
let rules = db.list_visibility_rules(&record.id).await.unwrap();
|
||||
if crate::visibility::listable_at_root(
|
||||
&rules,
|
||||
record.is_public,
|
||||
&record.owner_did,
|
||||
caller,
|
||||
) {
|
||||
expected.push(record.id.clone());
|
||||
}
|
||||
}
|
||||
expected.sort();
|
||||
let mut actual = db
|
||||
.list_visible_repos_page(caller, None, usize::MAX)
|
||||
.await
|
||||
.unwrap()
|
||||
.into_iter()
|
||||
.map(|record| record.id)
|
||||
.collect::<Vec<_>>();
|
||||
actual.sort();
|
||||
assert_eq!(actual, expected, "caller {caller:?}");
|
||||
}
|
||||
let schema = schema(db);
|
||||
for caller in [None, Some("did:key:zUnauthorized")] {
|
||||
let query = "{ repos { name } }";
|
||||
let response = match caller {
|
||||
None => anon(&schema, query).await,
|
||||
Some(caller) => authed(&schema, query, caller).await,
|
||||
};
|
||||
assert!(response.errors.is_empty(), "{:?}", response.errors);
|
||||
let mut names: Vec<String> = response.data.into_json().unwrap()["repos"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|repo| repo["name"].as_str().unwrap().to_owned())
|
||||
.collect();
|
||||
names.sort();
|
||||
assert_eq!(names, ["odd-star", "open", "subtree"]);
|
||||
}
|
||||
let query = "{ reposPage(limit: 1) { nodes { name ownerDid } hasNextPage endCursor } }";
|
||||
let response = anon(&schema, query).await;
|
||||
assert!(response.errors.is_empty(), "{:?}", response.errors);
|
||||
let json = response.data.into_json().unwrap();
|
||||
let page = &json["reposPage"];
|
||||
assert_eq!(
|
||||
page["nodes"],
|
||||
serde_json::json!([{"name": "odd-star", "ownerDid": OWNER}])
|
||||
);
|
||||
assert_eq!(page["hasNextPage"], true);
|
||||
let cursor = super::parse_repo_cursor(page["endCursor"].as_str().unwrap()).unwrap();
|
||||
assert_eq!(cursor, (OWNER.to_owned(), "odd-star".to_owned()));
|
||||
assert!(!json.to_string().contains("private"));
|
||||
assert!(!json.to_string().contains("quarantined"));
|
||||
|
||||
// Route-level check with unauthorized authenticated caller:
|
||||
// Excludes private, quarantined, and root-deny repos, matching anonymous behavior.
|
||||
let all_query = "{ reposPage(limit: 50) { nodes { name } hasNextPage } }";
|
||||
let unauth_response = authed(&schema, all_query, "did:key:zUnauthorized").await;
|
||||
assert!(
|
||||
unauth_response.errors.is_empty(),
|
||||
"{:?}",
|
||||
unauth_response.errors
|
||||
);
|
||||
let unauth_json = unauth_response.data.into_json().unwrap();
|
||||
let unauth_names: Vec<&str> = unauth_json["reposPage"]["nodes"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter_map(|n| n["name"].as_str())
|
||||
.collect();
|
||||
assert_eq!(unauth_names, vec!["odd-star", "open", "subtree"]);
|
||||
let unauth_str = unauth_json.to_string();
|
||||
assert!(!unauth_str.contains("private"));
|
||||
assert!(!unauth_str.contains("quarantined"));
|
||||
assert!(!unauth_str.contains("root-deny"));
|
||||
assert!(!unauth_str.contains("root-reader"));
|
||||
assert!(!unauth_str.contains("root-tie"));
|
||||
|
||||
// Reader caller gets root-reader and root-tie, but still excludes private, quarantined, root-deny.
|
||||
let reader_response = authed(&schema, all_query, reader).await;
|
||||
assert!(
|
||||
reader_response.errors.is_empty(),
|
||||
"{:?}",
|
||||
reader_response.errors
|
||||
);
|
||||
let reader_json = reader_response.data.into_json().unwrap();
|
||||
let reader_names: Vec<&str> = reader_json["reposPage"]["nodes"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter_map(|n| n["name"].as_str())
|
||||
.collect();
|
||||
assert_eq!(
|
||||
reader_names,
|
||||
vec!["odd-star", "open", "root-reader", "subtree"]
|
||||
);
|
||||
let reader_str = reader_json.to_string();
|
||||
assert!(!reader_str.contains("private"));
|
||||
assert!(!reader_str.contains("quarantined"));
|
||||
assert!(!reader_str.contains("root-deny"));
|
||||
assert!(!reader_str.contains("root-tie"));
|
||||
}
|
||||
|
||||
#[sqlx::test]
|
||||
async fn repos_page_rechecks_cursor_authority_and_exact_boundary(pool: PgPool) {
|
||||
let db = db(pool).await;
|
||||
let visible = repo("visible", OWNER, "a-visible", true);
|
||||
db.create_repo(&visible).await.unwrap();
|
||||
db.create_repo(&repo("hidden", OWNER, "z-private", false))
|
||||
.await
|
||||
.unwrap();
|
||||
let schema = schema(db);
|
||||
let first = anon(
|
||||
&schema,
|
||||
"{ reposPage(limit: 1) { nodes { name } hasNextPage endCursor } }",
|
||||
)
|
||||
.await;
|
||||
assert!(first.errors.is_empty());
|
||||
let first = first.data.into_json().unwrap();
|
||||
assert_eq!(first["reposPage"]["hasNextPage"], false);
|
||||
let query = format!(
|
||||
"{{ reposPage(limit: 1, after: \"{}\") {{ nodes {{ name }} hasNextPage endCursor }} }}",
|
||||
super::repo_cursor(&visible)
|
||||
);
|
||||
let owner = authed(&schema, &query, OWNER).await;
|
||||
assert!(owner.errors.is_empty());
|
||||
assert_eq!(
|
||||
owner.data.into_json().unwrap()["reposPage"]["nodes"][0]["name"],
|
||||
"z-private"
|
||||
);
|
||||
let anon = anon(&schema, &query).await;
|
||||
assert!(anon.errors.is_empty());
|
||||
assert_eq!(
|
||||
anon.data.into_json().unwrap()["reposPage"],
|
||||
serde_json::json!({
|
||||
"nodes": [], "hasNextPage": false, "endCursor": null
|
||||
})
|
||||
);
|
||||
let unauth = authed(&schema, &query, "did:key:zUnauthorized").await;
|
||||
assert!(unauth.errors.is_empty());
|
||||
assert_eq!(
|
||||
unauth.data.into_json().unwrap()["reposPage"],
|
||||
serde_json::json!({
|
||||
"nodes": [], "hasNextPage": false, "endCursor": null
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn repos_page_rejects_nul_cursors_before_database_access() {
|
||||
let pool = sqlx::postgres::PgPoolOptions::new()
|
||||
.connect_lazy("postgres://localhost/unused")
|
||||
.unwrap();
|
||||
let schema = schema(Arc::new(Db::for_testing(pool)));
|
||||
for position in [("did:key:zOwner\0", "repo"), ("did:key:zOwner", "repo\0")] {
|
||||
let cursor = super::URL_SAFE_NO_PAD.encode(serde_json::to_vec(&position).unwrap());
|
||||
let query = format!("{{ reposPage(after: \"{cursor}\") {{ hasNextPage }} }}");
|
||||
let response =
|
||||
tokio::time::timeout(std::time::Duration::from_secs(1), anon(&schema, &query))
|
||||
.await
|
||||
.expect("invalid cursors must be rejected without accessing the database");
|
||||
assert_eq!(response.errors.len(), 1);
|
||||
assert_eq!(response.errors[0].message, "invalid repository cursor");
|
||||
assert_eq!(response.data, async_graphql::Value::Null);
|
||||
}
|
||||
let position = ("did:key:zOwner", "repo");
|
||||
let cursor = super::URL_SAFE_NO_PAD.encode(serde_json::to_vec(&position).unwrap());
|
||||
assert_eq!(
|
||||
super::parse_repo_cursor(&cursor).unwrap(),
|
||||
(position.0.to_owned(), position.1.to_owned())
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn repos_page_rejects_invalid_inputs_before_database_access() {
|
||||
let pool = sqlx::postgres::PgPoolOptions::new()
|
||||
.connect_lazy("postgres://localhost/unused")
|
||||
.unwrap();
|
||||
let schema = schema(Arc::new(Db::for_testing(pool)));
|
||||
let over_max = crate::db::MAX_VISIBLE_REPO_PAGE_SIZE + 1;
|
||||
let expected_limit_message = format!(
|
||||
"limit must be between 1 and {}",
|
||||
crate::db::MAX_VISIBLE_REPO_PAGE_SIZE
|
||||
);
|
||||
// Each input pins its own rejection message: the loop must not accept
|
||||
// either message for every query, or a misattributed message (e.g. the
|
||||
// limit arm returning "invalid repository cursor") stays invisible.
|
||||
for (query, expected_message) in [
|
||||
(
|
||||
"{ reposPage(limit: 0) { hasNextPage } }".to_owned(),
|
||||
expected_limit_message.clone(),
|
||||
),
|
||||
(
|
||||
"{ reposPage(limit: -1) { hasNextPage } }".to_owned(),
|
||||
expected_limit_message.clone(),
|
||||
),
|
||||
(
|
||||
format!("{{ reposPage(limit: {over_max}) {{ hasNextPage }} }}"),
|
||||
expected_limit_message.clone(),
|
||||
),
|
||||
(
|
||||
"{ reposPage(after: \"invalid!\") { hasNextPage } }".to_owned(),
|
||||
"invalid repository cursor".to_owned(),
|
||||
),
|
||||
] {
|
||||
let response =
|
||||
tokio::time::timeout(std::time::Duration::from_secs(1), anon(&schema, &query))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(response.errors.len(), 1);
|
||||
assert_eq!(
|
||||
response.errors[0].message, expected_message,
|
||||
"wrong rejection message for query: {query}"
|
||||
);
|
||||
}
|
||||
// Valid base64 and valid cursor JSON: only the length guard rejects it.
|
||||
let oversized = super::URL_SAFE_NO_PAD
|
||||
.encode(serde_json::to_vec(&("did:key:reader", "a".repeat(3100))).unwrap());
|
||||
assert!(oversized.len() > 4096);
|
||||
assert!(serde_json::from_slice::<(String, String)>(
|
||||
&super::URL_SAFE_NO_PAD.decode(&oversized).unwrap()
|
||||
)
|
||||
.is_ok());
|
||||
assert!(super::parse_repo_cursor(&oversized).is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn parse_repo_cursor_rejects_wrong_shape_json() {
|
||||
// Valid base64 and valid JSON, but not a (String, String) pair: the
|
||||
// wrong-shape arm must report "invalid repository cursor", exactly,
|
||||
// before any database access.
|
||||
let pool = sqlx::postgres::PgPoolOptions::new()
|
||||
.connect_lazy("postgres://localhost/unused")
|
||||
.unwrap();
|
||||
let schema = schema(Arc::new(Db::for_testing(pool)));
|
||||
for payload in [
|
||||
serde_json::json!(["only-one"]),
|
||||
serde_json::json!(["a", "b", "c"]),
|
||||
serde_json::json!({"owner": "x", "name": "y"}),
|
||||
serde_json::json!(42),
|
||||
serde_json::json!(["a", 1]),
|
||||
] {
|
||||
let cursor = super::URL_SAFE_NO_PAD.encode(serde_json::to_vec(&payload).unwrap());
|
||||
let err = super::parse_repo_cursor(&cursor).unwrap_err();
|
||||
assert_eq!(err.message, "invalid repository cursor");
|
||||
let query = format!("{{ reposPage(after: \"{cursor}\") {{ hasNextPage }} }}");
|
||||
let response =
|
||||
tokio::time::timeout(std::time::Duration::from_secs(1), anon(&schema, &query))
|
||||
.await
|
||||
.expect("wrong-shape cursors must be rejected without accessing the database");
|
||||
assert_eq!(response.errors.len(), 1);
|
||||
assert_eq!(response.errors[0].message, "invalid repository cursor");
|
||||
assert_eq!(response.data, async_graphql::Value::Null);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn longest_repo_name_cursor_round_trips() {
|
||||
// #465 4th round: a cursor the server emits must always parse back.
|
||||
// Repo names are bounded at the create/fork/mirror write paths, so the
|
||||
// longest admissible name must produce a cursor under the 4096-byte
|
||||
// parse limit.
|
||||
let name = "a".repeat(crate::db::MAX_REPO_NAME_LEN);
|
||||
let record = repo("long", OWNER, &name, true);
|
||||
let cursor = super::repo_cursor(&record);
|
||||
assert!(
|
||||
cursor.len() <= 4096,
|
||||
"emitted cursor must stay under the parse limit: {}",
|
||||
cursor.len()
|
||||
);
|
||||
assert_eq!(
|
||||
super::parse_repo_cursor(&cursor).unwrap(),
|
||||
(OWNER.to_owned(), name)
|
||||
);
|
||||
}
|
||||
|
||||
/// A negative refUpdates(limit) must be accepted and clamped to zero rows,
|
||||
/// like tasks(limit: -1): the complexity formula's `limit.clamp(0, ...)`
|
||||
/// prices it at the floor and the shared collector returns before any
|
||||
/// database access. Mirrors tasks_negative_limit_clamped.
|
||||
#[tokio::test]
|
||||
async fn ref_updates_negative_limit_clamped() {
|
||||
let pool = sqlx::postgres::PgPoolOptions::new()
|
||||
.connect_lazy("postgres://localhost/unused")
|
||||
.unwrap();
|
||||
let schema = schema(Arc::new(Db::for_testing(pool)));
|
||||
let resp = tokio::time::timeout(
|
||||
std::time::Duration::from_secs(1),
|
||||
anon(&schema, "{ refUpdates(limit: -1) { repo } }"),
|
||||
)
|
||||
.await
|
||||
.expect("negative limit must clamp without accessing the database");
|
||||
assert!(
|
||||
resp.errors.is_empty(),
|
||||
"negative limit must clamp, not fail: {:?}",
|
||||
resp.errors
|
||||
);
|
||||
assert_eq!(count(&resp), 0);
|
||||
}
|
||||
|
||||
/// The mirror write path binds peer-supplied names verbatim, so it must
|
||||
/// enforce the same bound as the local write paths: an overlong peer name
|
||||
/// would otherwise make reposPage emit an unparseable cursor. The
|
||||
/// rejection happens before any database access.
|
||||
#[tokio::test]
|
||||
async fn mirror_repo_rejects_overlong_name() {
|
||||
let pool = sqlx::postgres::PgPoolOptions::new()
|
||||
.connect_lazy("postgres://localhost/unused")
|
||||
.unwrap();
|
||||
let db = Db::for_testing(pool);
|
||||
let name = "b".repeat(crate::db::MAX_REPO_NAME_LEN + 1);
|
||||
let err = db
|
||||
.upsert_mirror_repo("z6MkLong", &name, "/tmp/long", None, false)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
err.to_string().contains("exceeds"),
|
||||
"unexpected error: {err}"
|
||||
);
|
||||
}
|
||||
|
||||
async fn db(pool: PgPool) -> Arc<Db> {
|
||||
let db = Db::for_testing(pool);
|
||||
db.run_migrations().await.unwrap();
|
||||
@@ -475,12 +1211,13 @@ mod tests {
|
||||
assert_eq!(count_tasks(&resp), 0);
|
||||
}
|
||||
|
||||
/// #255: ceiling of the tasks(limit) clamp must be held (schema promises Max 200).
|
||||
/// #255: ceiling of the tasks(limit) clamp must be held (schema promises the
|
||||
/// documented maximum).
|
||||
#[sqlx::test]
|
||||
async fn tasks_limit_ceiling_clamped_to_200(pool: PgPool) {
|
||||
async fn tasks_limit_ceiling_clamped_to_max(pool: PgPool) {
|
||||
let db = db(pool).await;
|
||||
let now = Utc::now().to_rfc3339();
|
||||
for i in 0..201 {
|
||||
for i in 0..crate::db::MAX_VISIBLE_TASK_PAGE_SIZE + 1 {
|
||||
db.create_task(&crate::db::AgentTask {
|
||||
id: format!("task-ceil-{i}"),
|
||||
repo_id: None,
|
||||
@@ -501,7 +1238,49 @@ mod tests {
|
||||
}
|
||||
let schema = schema(db);
|
||||
let resp = anon(&schema, "{ tasks(limit: 5000) { id } }").await;
|
||||
assert_eq!(count_tasks(&resp), 200, "limit above 200 must clamp to 200");
|
||||
assert_eq!(
|
||||
count_tasks(&resp),
|
||||
crate::db::MAX_VISIBLE_TASK_PAGE_SIZE as usize,
|
||||
"limit above the maximum must clamp to MAX_VISIBLE_TASK_PAGE_SIZE"
|
||||
);
|
||||
}
|
||||
|
||||
/// A limit above MAX_VISIBLE_REF_UPDATES must reach the resolver clamped,
|
||||
/// not be rejected: the complexity formula clamps the priced limit (so the
|
||||
/// document passes validation) and the shared collector clamps the rows
|
||||
/// (so at most MAX_VISIBLE_REF_UPDATES come back). Removing the formula
|
||||
/// clamp prices the raw limit and rejects the document; removing the
|
||||
/// collector clamp returns unbounded rows. Mirrors
|
||||
/// tasks_limit_ceiling_clamped_to_max.
|
||||
#[sqlx::test]
|
||||
async fn ref_updates_limit_ceiling_clamped_to_max(pool: PgPool) {
|
||||
use crate::api::events::MAX_VISIBLE_REF_UPDATES;
|
||||
let db = db(pool).await;
|
||||
db.create_repo(&repo("r1", OWNER, "widget", true))
|
||||
.await
|
||||
.unwrap();
|
||||
let total = MAX_VISIBLE_REF_UPDATES as usize + 1;
|
||||
for i in 0..total {
|
||||
let mut row = ref_row(&format!("ceil{i}"), "z6MkOwner/widget");
|
||||
row.timestamp = format!("2026-07-01T10:{:02}:{:02}+00:00", i / 60, i % 60);
|
||||
db.insert_ref_update(&row).await.unwrap();
|
||||
}
|
||||
let schema = schema(db);
|
||||
let query = format!(
|
||||
"{{ refUpdates(limit: {}) {{ repo }} }}",
|
||||
MAX_VISIBLE_REF_UPDATES + 4800
|
||||
);
|
||||
let resp = anon(&schema, &query).await;
|
||||
assert!(
|
||||
resp.errors.is_empty(),
|
||||
"limit above the maximum must clamp, not fail: {:?}",
|
||||
resp.errors
|
||||
);
|
||||
assert_eq!(
|
||||
count(&resp),
|
||||
MAX_VISIBLE_REF_UPDATES as usize,
|
||||
"limit above the maximum must clamp to MAX_VISIBLE_REF_UPDATES"
|
||||
);
|
||||
}
|
||||
|
||||
fn count_tasks(resp: &async_graphql::Response) -> usize {
|
||||
|
||||
@@ -20,6 +20,7 @@ impl SubscriptionRoot {
|
||||
/// `if announce` block, `api/repos.rs`). This is a single-point invariant —
|
||||
/// any new sender to `ref_update_tx` MUST be `announce`-gated, or private-repo
|
||||
/// ref metadata leaks here to unauthenticated subscribers (#112/#114 class).
|
||||
#[graphql(complexity = "50 + child_complexity")]
|
||||
async fn ref_updates(
|
||||
&self,
|
||||
ctx: &Context<'_>,
|
||||
@@ -46,6 +47,7 @@ impl SubscriptionRoot {
|
||||
})
|
||||
}
|
||||
|
||||
#[graphql(complexity = "50 + child_complexity")]
|
||||
async fn task_events(
|
||||
&self,
|
||||
ctx: &Context<'_>,
|
||||
|
||||
@@ -11,6 +11,14 @@ pub struct RepoType {
|
||||
pub created_at: String,
|
||||
}
|
||||
|
||||
#[derive(SimpleObject)]
|
||||
pub struct RepoPageType {
|
||||
pub nodes: Vec<RepoType>,
|
||||
pub has_next_page: bool,
|
||||
/// Position after the last returned visible repository; null for an empty page.
|
||||
pub end_cursor: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(SimpleObject, Clone)]
|
||||
pub struct AgentTaskType {
|
||||
pub id: String,
|
||||
|
||||
@@ -2638,6 +2638,7 @@ mod tests {
|
||||
/// type probe carries its oid on stdin rather than in argv, so an oid appears in
|
||||
/// the log only once an object has already got past its probe, and a healthy
|
||||
/// object costs two invocations to a faulting one's one.
|
||||
#[cfg(unix)]
|
||||
fn objects_attempted(log: &std::path::Path) -> usize {
|
||||
std::fs::read_to_string(log)
|
||||
.unwrap_or_default()
|
||||
|
||||
@@ -912,6 +912,7 @@ mod tests {
|
||||
/// probe carries its oid on stdin rather than in argv, so an oid appears in the log only
|
||||
/// once an object has already got past its probe, and a healthy object costs two
|
||||
/// invocations to a faulting one's one.
|
||||
#[cfg(unix)]
|
||||
fn objects_attempted(log: &std::path::Path) -> usize {
|
||||
std::fs::read_to_string(log)
|
||||
.unwrap_or_default()
|
||||
|
||||
@@ -1330,6 +1330,18 @@ mod tests {
|
||||
.await;
|
||||
}
|
||||
|
||||
/// Use a rooted path without a Windows drive prefix in the remote fixture.
|
||||
/// The destination still resolves on the temp directory's current drive,
|
||||
/// while the remote's relative path contains no illegal colon component.
|
||||
fn absolute_slug_path(path: &Path) -> String {
|
||||
assert!(path.is_absolute());
|
||||
path.components()
|
||||
.filter(|part| !matches!(part, std::path::Component::Prefix(_)))
|
||||
.collect::<std::path::PathBuf>()
|
||||
.to_string_lossy()
|
||||
.replace('\\', "/")
|
||||
}
|
||||
|
||||
#[sqlx::test]
|
||||
async fn process_batch_rejects_slug_escaping_repos_dir(pool: PgPool) {
|
||||
// The verified escape from #272: `PathBuf::join` discards everything
|
||||
@@ -1345,12 +1357,13 @@ mod tests {
|
||||
// fails on but leaves the parent it created, so assert on both.
|
||||
let outside = TempDir::new().unwrap();
|
||||
let escape_dir = outside.path().join("nest");
|
||||
let slug = format!("a/{}/escape", escape_dir.display());
|
||||
let escape_path = absolute_slug_path(&escape_dir);
|
||||
let slug = format!("a/{escape_path}/escape");
|
||||
let escape_target = escape_dir.join("escape.git");
|
||||
|
||||
// Serve the composed URL for real, so a run without the guard genuinely
|
||||
// clones outside the root rather than merely failing at git.
|
||||
let rel = format!("a{}/escape", escape_dir.display());
|
||||
let rel = format!("a{escape_path}/escape");
|
||||
let (_remote, peer_url) = rooted_remote(&[&rel]);
|
||||
|
||||
let did = "did:key:z6MkAttacker";
|
||||
@@ -1470,6 +1483,7 @@ mod tests {
|
||||
// ── canonical containment before the git call (issue #272) ───────────────
|
||||
|
||||
/// Every ref in `repo`, as one string, for a before/after comparison.
|
||||
#[cfg(unix)]
|
||||
fn refs_of(repo: &Path) -> String {
|
||||
let out = Command::new("git")
|
||||
.args(["-C", repo.to_str().unwrap(), "for-each-ref"])
|
||||
@@ -1802,6 +1816,7 @@ mod tests {
|
||||
/// starvation tests is fixed rather than dependent on how fast the loop
|
||||
/// runs. Two rows enqueued in the same microsecond would otherwise order
|
||||
/// arbitrarily.
|
||||
#[cfg(unix)]
|
||||
async fn enqueue_at(db: &Db, pool: &PgPool, repo: &str, did: &str, enqueued_at: &str) {
|
||||
enqueue(db, repo, did).await;
|
||||
sqlx::query("UPDATE sync_queue SET enqueued_at = $1 WHERE repo = $2")
|
||||
@@ -1995,7 +2010,8 @@ mod tests {
|
||||
|
||||
let outside = TempDir::new().unwrap();
|
||||
let escape_dir = outside.path().join("nest");
|
||||
let slug = format!("a/{}/gitlawb-probe", escape_dir.display());
|
||||
let escape_path = absolute_slug_path(&escape_dir);
|
||||
let slug = format!("a/{escape_path}/gitlawb-probe");
|
||||
let escape_target = escape_dir.join("gitlawb-probe.git");
|
||||
|
||||
// Two things this fixture must get right or the test is green for the
|
||||
@@ -2006,7 +2022,7 @@ mod tests {
|
||||
// run without the guard genuinely clones outside the root instead of
|
||||
// just failing at git. Db::upsert_peer cannot seed this row: it gates on
|
||||
// is_public_http_url, which rejects file://.
|
||||
let rel = format!("a{}/gitlawb-probe", escape_dir.display());
|
||||
let rel = format!("a{escape_path}/gitlawb-probe");
|
||||
let (_remote, peer_url) = rooted_remote(&[&rel]);
|
||||
let did = "did:key:z6MkAttacker";
|
||||
seed_local_peer(&pool, did, &peer_url).await;
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
//! Native Git fixtures shared by timing tests on Unix and Windows.
|
||||
|
||||
use std::ops::Deref;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::OnceLock;
|
||||
|
||||
/// The fixture delays or hangs in its own process, so the production watchdog
|
||||
/// can terminate it without depending on a shell or descendant process support.
|
||||
pub(super) enum Behavior<'a> {
|
||||
Delay(u64),
|
||||
Hang,
|
||||
HangOid(&'a str),
|
||||
HangRepo(&'a str),
|
||||
LogTypes(&'a Path),
|
||||
}
|
||||
|
||||
pub(super) struct GitShim {
|
||||
_directory: tempfile::TempDir,
|
||||
executable: PathBuf,
|
||||
}
|
||||
|
||||
impl Deref for GitShim {
|
||||
type Target = Path;
|
||||
|
||||
fn deref(&self) -> &Path {
|
||||
&self.executable
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn create(name: &str, behavior: Behavior<'_>) -> GitShim {
|
||||
static COMPILED: OnceLock<tempfile::TempDir> = OnceLock::new();
|
||||
let compiled = COMPILED.get_or_init(|| {
|
||||
let directory = tempfile::tempdir().expect("native Git fixture directory");
|
||||
let source = directory.path().join("git_shim.rs");
|
||||
std::fs::write(&source, include_str!("../tests/fixtures/git_shim.rs"))
|
||||
.expect("write native Git fixture source");
|
||||
let executable = directory
|
||||
.path()
|
||||
.join(format!("git-shim{}", std::env::consts::EXE_SUFFIX));
|
||||
let output =
|
||||
std::process::Command::new(std::env::var_os("RUSTC").unwrap_or_else(|| "rustc".into()))
|
||||
.arg("--edition=2021")
|
||||
.arg("-Dwarnings")
|
||||
.arg(&source)
|
||||
.arg("-o")
|
||||
.arg(executable)
|
||||
.output()
|
||||
.expect("compile native Git fixture with the installed Rust toolchain");
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"{}",
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
directory
|
||||
});
|
||||
let directory = tempfile::Builder::new()
|
||||
.prefix(name)
|
||||
.tempdir()
|
||||
.expect("Git fixture instance");
|
||||
let file_name = format!("git-shim{}", std::env::consts::EXE_SUFFIX);
|
||||
let executable = directory.path().join(&file_name);
|
||||
std::fs::copy(compiled.path().join(file_name), &executable).expect("copy native Git fixture");
|
||||
let config = match behavior {
|
||||
Behavior::Delay(milliseconds) => format!("delay\n{milliseconds}"),
|
||||
Behavior::Hang => "hang\n".to_owned(),
|
||||
Behavior::HangOid(oid) => format!("hang-oid\n{oid}"),
|
||||
Behavior::HangRepo(repo) => format!("hang-repo\n{repo}"),
|
||||
Behavior::LogTypes(path) => format!("log-types\n{}", path.display()),
|
||||
};
|
||||
std::fs::write(directory.path().join("config"), config).expect("configure native Git fixture");
|
||||
GitShim {
|
||||
_directory: directory,
|
||||
executable,
|
||||
}
|
||||
}
|
||||
@@ -28,6 +28,9 @@ use gitlawb_core::identity::Keypair;
|
||||
use crate::auth::AuthenticatedDid;
|
||||
use crate::state::AppState;
|
||||
|
||||
#[path = "test_git_shim.rs"]
|
||||
mod git_shim;
|
||||
|
||||
/// Build an [`AppState`] over a real, migrated Postgres pool (from `#[sqlx::test]`).
|
||||
/// Runs the schema migrations first, because the per-test database starts empty.
|
||||
///
|
||||
@@ -559,11 +562,12 @@ mod tests {
|
||||
/// #174 (SC1, load-bearing): a saturated READ pool must NOT shed an
|
||||
/// authenticated push — the write pool is a separate budget. Read pool at zero,
|
||||
/// write pool with capacity: the push proceeds PAST admission (it then errors on
|
||||
/// the placeholder DB, but crucially it is not a 503). Route git-receive-pack
|
||||
/// the closed DB, with db_unavailable rather than overloaded). Route git-receive-pack
|
||||
/// back to the read pool and this goes red — that is the isolation proof.
|
||||
#[tokio::test]
|
||||
async fn git_receive_pack_not_shed_by_exhausted_read_pool() {
|
||||
let mut state = test_state_lazy();
|
||||
state.db.pool().close().await;
|
||||
// Read pool exhausted as if a flood of anonymous clones held every slot.
|
||||
state.git_read_semaphore = Arc::new(tokio::sync::Semaphore::new(0));
|
||||
// Write pool keeps its default capacity from test_state_lazy.
|
||||
@@ -585,10 +589,13 @@ mod tests {
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_ne!(
|
||||
resp.status(),
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"an exhausted READ pool must not shed a push — the write pool is a separate budget (#174)"
|
||||
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
|
||||
let bytes = axum::body::to_bytes(resp.into_body(), 1024).await.unwrap();
|
||||
let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
|
||||
assert_eq!(
|
||||
body["error"],
|
||||
crate::error::DB_UNAVAILABLE_CODE,
|
||||
"the push must clear admission and reach the deliberately closed database"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -11195,14 +11202,10 @@ mod tests {
|
||||
seed_legacy_pin(&pool, &bare, &fx.public_oid, Some(&repo.id)).await;
|
||||
|
||||
// A git that takes 300ms per invocation (the read makes two: type, then content).
|
||||
let slow_git = std::env::temp_dir().join(format!("gl-slow-git-{short}"));
|
||||
std::fs::write(&slow_git, "#!/bin/sh\nsleep 0.3\nexec git \"$@\"\n").expect("write shim");
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
std::fs::set_permissions(&slow_git, std::fs::Permissions::from_mode(0o755))
|
||||
.expect("chmod shim");
|
||||
}
|
||||
let slow_git = git_shim::create(
|
||||
&format!("gl-slow-git-{short}"),
|
||||
git_shim::Behavior::Delay(300),
|
||||
);
|
||||
|
||||
let ticks = std::sync::Arc::new(AtomicUsize::new(0));
|
||||
let ticker = {
|
||||
@@ -12149,19 +12152,6 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// Write an executable `git` stand-in and return its path.
|
||||
fn write_git_shim(name: &str, script: &str) -> std::path::PathBuf {
|
||||
let path = std::env::temp_dir().join(name);
|
||||
std::fs::write(&path, script).expect("write the git shim");
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755))
|
||||
.expect("chmod the git shim");
|
||||
}
|
||||
path
|
||||
}
|
||||
|
||||
/// F6 scenario 1 (#173 round 13): one hung candidate must not starve the rows behind
|
||||
/// it in the same pass. `DiscoveryCtx` is loaded once per pass, so before the per-row
|
||||
/// slice every source-less row in a pass shared ONE deadline: the first row's wedged
|
||||
@@ -12201,23 +12191,9 @@ mod tests {
|
||||
|
||||
// The type stage feeds the oid on STDIN (`cat-file --batch-check`) and the
|
||||
// content stage puts it in argv, so the stand-in has to look in both places.
|
||||
let git_bin = write_git_shim(
|
||||
let git_bin = git_shim::create(
|
||||
&format!("gl-hung-git-{short}"),
|
||||
&format!(
|
||||
"#!/bin/sh\n\
|
||||
if [ \"$2\" = \"--batch-check\" ]; then\n\
|
||||
\x20 oid=$(cat)\n\
|
||||
\x20 case \"$oid\" in\n\
|
||||
\x20 {hung_oid}) sleep 30; exit 1 ;;\n\
|
||||
\x20 esac\n\
|
||||
\x20 printf '%s\\n' \"$oid\" | git \"$@\"\n\
|
||||
\x20 exit $?\n\
|
||||
fi\n\
|
||||
case \"$*\" in\n\
|
||||
\x20 *{hung_oid}*) sleep 30; exit 1 ;;\n\
|
||||
esac\n\
|
||||
exec git \"$@\"\n"
|
||||
),
|
||||
git_shim::Behavior::HangOid(&hung_oid),
|
||||
);
|
||||
|
||||
let stats = tokio::time::timeout(
|
||||
@@ -12302,10 +12278,7 @@ mod tests {
|
||||
|
||||
// Wedges on every invocation, so no row can ever be repaired and the only
|
||||
// question left is what each one COSTS.
|
||||
let git_bin = write_git_shim(
|
||||
&format!("gl-spent-git-{short}"),
|
||||
"#!/bin/sh\nsleep 30\nexit 1\n",
|
||||
);
|
||||
let git_bin = git_shim::create(&format!("gl-spent-git-{short}"), git_shim::Behavior::Hang);
|
||||
|
||||
let stats = tokio::time::timeout(
|
||||
std::time::Duration::from_secs(60),
|
||||
@@ -12909,10 +12882,7 @@ mod tests {
|
||||
seed_legacy_pin(&pool, &src, oid, None).await;
|
||||
}
|
||||
|
||||
let git_bin = write_git_shim(
|
||||
&format!("gl-starve-git-{short}"),
|
||||
"#!/bin/sh\nsleep 30\nexit 1\n",
|
||||
);
|
||||
let git_bin = git_shim::create(&format!("gl-starve-git-{short}"), git_shim::Behavior::Hang);
|
||||
|
||||
tokio::time::timeout(
|
||||
std::time::Duration::from_secs(120),
|
||||
@@ -12980,9 +12950,9 @@ mod tests {
|
||||
let (raw_cid, provider_cid) = seed_legacy_pin(&pool, &src, &fx.public_oid, None).await;
|
||||
|
||||
// Wedges only inside the position-nine repo, which the sweep enters by cwd.
|
||||
let git_bin = write_git_shim(
|
||||
let git_bin = git_shim::create(
|
||||
&format!("gl-mid-git-{short}"),
|
||||
"#!/bin/sh\ncase \"$(pwd)\" in\n */midcand9.git) sleep 30; exit 1 ;;\nesac\nexec git \"$@\"\n",
|
||||
git_shim::Behavior::HangRepo("midcand9.git"),
|
||||
);
|
||||
|
||||
let first = tokio::time::timeout(
|
||||
@@ -13144,19 +13114,9 @@ mod tests {
|
||||
|
||||
let log = std::env::temp_dir().join(format!("gl-ali-log-{short}"));
|
||||
let _ = std::fs::remove_file(&log);
|
||||
let git_bin = write_git_shim(
|
||||
let git_bin = git_shim::create(
|
||||
&format!("gl-ali-git-{short}"),
|
||||
&format!(
|
||||
"#!/bin/sh\n\
|
||||
if [ \"$2\" = \"--batch-check\" ]; then\n\
|
||||
\x20 oid=$(cat)\n\
|
||||
\x20 printf '%s %s\\n' \"$oid\" \"$(basename $(pwd))\" >> {log}\n\
|
||||
\x20 printf '%s\\n' \"$oid\" | git \"$@\"\n\
|
||||
\x20 exit $?\n\
|
||||
fi\n\
|
||||
exec git \"$@\"\n",
|
||||
log = log.display()
|
||||
),
|
||||
git_shim::Behavior::LogTypes(&log),
|
||||
);
|
||||
|
||||
let mut traversal = crate::ipfs_pin::DiscoveryTraversalState::default();
|
||||
|
||||
+59
@@ -0,0 +1,59 @@
|
||||
use std::io::{Read, Write};
|
||||
use std::process::{Command, Stdio};
|
||||
use std::time::Duration;
|
||||
|
||||
fn main() -> std::io::Result<()> {
|
||||
let executable = std::env::current_exe()?;
|
||||
let config = std::fs::read_to_string(executable.parent().unwrap().join("config"))?;
|
||||
let (mode, parameter) = config.split_once('\n').unwrap_or((&config, ""));
|
||||
let args: Vec<_> = std::env::args_os().skip(1).collect();
|
||||
if mode == "delay" {
|
||||
std::thread::sleep(Duration::from_millis(parameter.parse().unwrap()));
|
||||
}
|
||||
let batch = args.get(1).is_some_and(|arg| arg == "--batch-check");
|
||||
let input = if batch {
|
||||
let mut bytes = Vec::new();
|
||||
std::io::stdin().read_to_end(&mut bytes)?;
|
||||
Some(bytes)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let cwd = std::env::current_dir()?;
|
||||
let hang = mode == "hang"
|
||||
|| (mode == "hang-repo" && cwd.file_name().is_some_and(|name| name == parameter))
|
||||
|| (mode == "hang-oid"
|
||||
&& (args.iter().any(|arg| arg == parameter)
|
||||
|| input
|
||||
.as_ref()
|
||||
.is_some_and(|bytes| String::from_utf8_lossy(bytes).trim() == parameter)));
|
||||
if hang {
|
||||
std::thread::sleep(Duration::from_secs(30));
|
||||
std::process::exit(1);
|
||||
}
|
||||
if mode == "log-types" && batch {
|
||||
let mut log = std::fs::OpenOptions::new()
|
||||
.create(true)
|
||||
.append(true)
|
||||
.open(parameter)?;
|
||||
writeln!(
|
||||
log,
|
||||
"{} {}",
|
||||
String::from_utf8_lossy(input.as_ref().unwrap()).trim(),
|
||||
cwd.file_name().unwrap().to_string_lossy()
|
||||
)?;
|
||||
}
|
||||
let mut child = Command::new("git")
|
||||
.args(&args)
|
||||
.stdin(if input.is_some() {
|
||||
Stdio::piped()
|
||||
} else {
|
||||
Stdio::inherit()
|
||||
})
|
||||
.stdout(Stdio::inherit())
|
||||
.stderr(Stdio::inherit())
|
||||
.spawn()?;
|
||||
if let Some(bytes) = input {
|
||||
child.stdin.take().unwrap().write_all(&bytes)?;
|
||||
}
|
||||
std::process::exit(child.wait()?.code().unwrap_or(1));
|
||||
}
|
||||
@@ -8,6 +8,7 @@ Step-by-step guide to staking $GITLAWB, registering your node on-chain, and earn
|
||||
|
||||
- A wallet with at least **10,000 $GITLAWB** (minimum stake) plus a small amount of ETH on Base for gas
|
||||
- Docker or Rust 1.91+ (for running the node process)
|
||||
- PostgreSQL 16+ — the node uses `pg_input_is_valid`, which exists only on PostgreSQL 16 and later. The bundled `docker-compose.yml` pins `postgres:16-alpine`; if you supply your own `DATABASE_URL`, the server must be 16+.
|
||||
- A public HTTP URL (your-host.com) — can be a VPS, Fly.io app, or anything reachable. A Fly.io config is provided at `infra/fly/fly.toml` (deploy from the repo root with `fly deploy -c infra/fly/fly.toml`)
|
||||
|
||||
---
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
# GraphQL repository pagination
|
||||
|
||||
`repos` returns the complete visible repository list when it contains at most
|
||||
200 entries. Larger lists return a GraphQL error directing the caller to
|
||||
`reposPage`; they are never silently truncated.
|
||||
|
||||
Use `reposPage` for larger lists:
|
||||
|
||||
```graphql
|
||||
query Repositories($after: String) {
|
||||
reposPage(limit: 50, after: $after) {
|
||||
nodes {
|
||||
name
|
||||
ownerDid
|
||||
}
|
||||
hasNextPage
|
||||
endCursor
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Start with `after: null`. When `hasNextPage` is true, pass the returned
|
||||
`endCursor` as `after` on the next request. Stop when `hasNextPage` is false.
|
||||
An empty page has a null `endCursor`. The default limit is 50 and the maximum
|
||||
page size is 200; limits below 1 or above 200 are rejected (see Query budgets
|
||||
below). Malformed cursors, including decoded owner or name strings containing
|
||||
NUL, return `invalid repository cursor` before database access.
|
||||
|
||||
Pages are ordered by normalized owner DID and repository name. Visibility and
|
||||
mirror deduplication are applied in the database before limiting the page;
|
||||
hidden and quarantined repositories do not occupy page slots or create a
|
||||
continuation signal. Cursors contain a position from the last returned visible
|
||||
repository, not a permission grant. Each request checks the current caller's
|
||||
visibility independently. Keep the same caller while traversing a list.
|
||||
Malformed reader lists in a repository's root visibility rule deny access to
|
||||
non-owners for that repository while other visible repositories remain listable.
|
||||
|
||||
Pagination is not a snapshot: concurrent renames, ownership changes, or visibility
|
||||
changes can alter subsequent pages. Treat cursors as opaque and restart from the
|
||||
first page when a fresh complete traversal is needed. Small legacy `repos`
|
||||
queries retain their activity ordering.
|
||||
|
||||
## Query budgets
|
||||
|
||||
Every GraphQL document is validated against two budgets before any resolver
|
||||
runs. Both apply per document (per request), not per client: `/graphql` has
|
||||
no per-caller rate limiting, so a rejected document can always be retried
|
||||
smaller. The constants live in `crates/gitlawb-node/src/graphql/mod.rs`.
|
||||
|
||||
- **Complexity budget: 400 per document** (`GRAPHQL_MAX_COMPLEXITY`). Each
|
||||
root field carries a base cost plus the clamped page size plus the selected
|
||||
child fields:
|
||||
- `repos`: 50 + `MAX_VISIBLE_REPO_PAGE_SIZE` + child fields. `repos`
|
||||
always scans the full visible corpus, so it is priced as the full page
|
||||
it is.
|
||||
- `reposPage`: 50 + `MAX_VISIBLE_REPO_PAGE_SIZE` + child fields. Every
|
||||
page request sorts the full deduplicated corpus, so the price does not
|
||||
depend on `limit`: seven `reposPage(limit: 1)` aliases cost the same as
|
||||
seven max-size pages.
|
||||
- `refUpdates`: 100 + clamped `limit` (`MAX_VISIBLE_REF_UPDATES`) + child
|
||||
fields. The 100 base covers the collector's fixed scan (the deduplicated
|
||||
repo set, the quarantined set, the visibility rules, and the bounded
|
||||
up-to-2048-row walk past withheld rows); the limit term prices only the
|
||||
rows returned.
|
||||
- `tasks`: 50 + clamped `limit` (`MAX_VISIBLE_TASK_PAGE_SIZE`) + child
|
||||
fields.
|
||||
- `task`, mutations, subscriptions: 50 + child fields.
|
||||
|
||||
A document whose total exceeds 400 is rejected with
|
||||
`Query is too complex.` before any resolver runs, so an over-budget
|
||||
document needs fewer aliases or fewer child fields. A smaller `limit` does
|
||||
not help `reposPage` or `repos`: both are priced at the full
|
||||
`MAX_VISIBLE_REPO_PAGE_SIZE` scan regardless of `limit`.
|
||||
|
||||
- **Depth cap: 14 per document** (`GRAPHQL_MAX_DEPTH`). Documents nested
|
||||
deeper than 14 selections are rejected with `Query is nested too deep.`
|
||||
Reference in new issue
Block a user