From 1b07caf646cb3ecb57e44f637f8d5d8764c12993 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Tue, 3 Nov 2026 01:53:52 -0500 Subject: [PATCH] wai-quantum-service: a job the service was running when it stopped runs once more A job that was running when the service stopped came back interrupted, its request kept, and nothing ran it again unless its client sent it again. Since the engine gives the same result for the same request, the service now runs it again itself: on start, a job that was running is queued again in its order, to run once more from its stored request, and its `interruptions` say when the service stopped. Stopped again while it runs once more, it comes back interrupted with the reason and is not run a third time (RERUNS = 1), so a job that stops the service whenever it runs does not run for ever. GET /v1's `restored` counts the jobs run again. wai.quantum.job/1 now lists `interrupted` among its statuses, which the service has returned since its jobs were kept across restarts, and the optional `interruptions`. Also: the store's pseudonym test held the hex of a keyed hash to not containing "203", three digits of the address, which a random key gives about once in 140 runs. It now holds it to hex, which cannot spell the address. Co-Authored-By: Claude Opus 5.5 --- wai/CHANGELOG.md | 15 ++ wai/wai-quantum-cli/docs/llms-full.txt | 2 +- wai/wai-quantum-cli/docs/local-mcp.json | 45 ++++-- wai/wai-quantum-cli/docs/service-mcp.json | 62 ++++++-- wai/wai-quantum-service/README.md | 5 +- wai/wai-quantum-service/docs/provenance.md | 2 +- wai/wai-quantum-service/docs/self-host.md | 2 +- wai/wai-quantum-service/src/discovery.rs | 3 +- wai/wai-quantum-service/src/http.rs | 2 +- wai/wai-quantum-service/src/jobs.rs | 177 ++++++++++++++++++--- wai/wai-quantum-service/src/main.rs | 3 +- wai/wai-quantum-service/src/mcp.rs | 2 +- wai/wai-quantum-service/src/store.rs | 6 +- wai/wai-quantum-service/tests/restart.rs | 36 ++++- wai/wai-quantum/schemas/job.v1.json | 5 +- wai/wai-quantum/src/engine/schema.rs | 2 + 16 files changed, 305 insertions(+), 64 deletions(-) diff --git a/wai/CHANGELOG.md b/wai/CHANGELOG.md index 4cd7a06b5..67b0031d6 100644 --- a/wai/CHANGELOG.md +++ b/wai/CHANGELOG.md @@ -416,6 +416,21 @@ limits. The three MCP tools that run also declared `idempotentHint: true`, which - **wai-quantum for Python:** `Service.submit` and `Service.run` do the same, and take `idempotency_key`. A job the service interrupted now ends the wait; before, the wait went on to its deadline. +### The job service runs a job it was running when it stopped, once more + +A job that was running when the service stopped came back `interrupted`, its request kept. Nothing ran it again unless +its client sent it again. + +- **wai-quantum-service:** + - On start, a job that was running is queued again, in its order, to run once more from its stored request. Its + `interruptions` say when the service stopped. + - If the service stops again while the job runs once more, the job comes back `interrupted`, with the reason, and is + not run a third time. A job that stops the service whenever it runs therefore does not run for ever. + - `GET /v1`'s `restored` counts the jobs run again (`rerun`). +- **wai-quantum:** `wai.quantum.job/1` lists `interrupted` among its statuses. The service has returned that status + since its jobs were first kept across restarts, but the schema did not list it. The schema also gains the optional + `interruptions`. + ### The receipt ledger holds a decode to the frames it delivered, as one animated PNG - **Tariff and invoice version 5** ([extensions/receipt-ledger.md](extensions/receipt-ledger.md) §4.8): version 4's diff --git a/wai/wai-quantum-cli/docs/llms-full.txt b/wai/wai-quantum-cli/docs/llms-full.txt index 672770549..e9b8aad15 100644 --- a/wai/wai-quantum-cli/docs/llms-full.txt +++ b/wai/wai-quantum-cli/docs/llms-full.txt @@ -696,7 +696,7 @@ Parameters: ### `get_quantum_job`: Read a quantum job -A job's status, and its result when it is done. The service keeps every job across its restarts for its retention (a day on the public service); a job that was running when it stopped comes back interrupted, with the reason. +A job's status, and its result when it is done. The service keeps every job across its restarts for its retention (a day on the public service); a job that was running when it stopped is run again once from its request (its interruptions say when), and stopped again it comes back interrupted, with the reason. Parameters: - `id` (string, required) diff --git a/wai/wai-quantum-cli/docs/local-mcp.json b/wai/wai-quantum-cli/docs/local-mcp.json index b167d774c..c2fa94a3e 100644 --- a/wai/wai-quantum-cli/docs/local-mcp.json +++ b/wai/wai-quantum-cli/docs/local-mcp.json @@ -558,8 +558,10 @@ "running", "done", "failed", - "cancelled" - ] + "cancelled", + "interrupted" + ], + "description": "interrupted: the job service stopped while the job ran, after it had run it again once from its request, so it has no result; error says so." }, "method": { "type": "string", @@ -7633,7 +7635,14 @@ }, "error": { "type": "string", - "description": "Why a job failed." + "description": "Why a job failed, or why it was interrupted." + }, + "interruptions": { + "type": "array", + "items": { + "type": "integer" + }, + "description": "The job service: each time it stopped while the job ran, in milliseconds since 1970. A job that was running when the service stopped is queued again, once, to run from its stored request." }, "submitted_ms": { "type": "integer", @@ -30353,8 +30362,10 @@ "running", "done", "failed", - "cancelled" - ] + "cancelled", + "interrupted" + ], + "description": "interrupted: the job service stopped while the job ran, after it had run it again once from its request, so it has no result; error says so." }, "method": { "type": "string", @@ -37428,7 +37439,14 @@ }, "error": { "type": "string", - "description": "Why a job failed." + "description": "Why a job failed, or why it was interrupted." + }, + "interruptions": { + "type": "array", + "items": { + "type": "integer" + }, + "description": "The job service: each time it stopped while the job ran, in milliseconds since 1970. A job that was running when the service stopped is queued again, once, to run from its stored request." }, "submitted_ms": { "type": "integer", @@ -49850,8 +49868,10 @@ "running", "done", "failed", - "cancelled" - ] + "cancelled", + "interrupted" + ], + "description": "interrupted: the job service stopped while the job ran, after it had run it again once from its request, so it has no result; error says so." }, "method": { "type": "string", @@ -56925,7 +56945,14 @@ }, "error": { "type": "string", - "description": "Why a job failed." + "description": "Why a job failed, or why it was interrupted." + }, + "interruptions": { + "type": "array", + "items": { + "type": "integer" + }, + "description": "The job service: each time it stopped while the job ran, in milliseconds since 1970. A job that was running when the service stopped is queued again, once, to run from its stored request." }, "submitted_ms": { "type": "integer", diff --git a/wai/wai-quantum-cli/docs/service-mcp.json b/wai/wai-quantum-cli/docs/service-mcp.json index 83d3690ff..d0b0148ed 100644 --- a/wai/wai-quantum-cli/docs/service-mcp.json +++ b/wai/wai-quantum-cli/docs/service-mcp.json @@ -571,8 +571,10 @@ "running", "done", "failed", - "cancelled" - ] + "cancelled", + "interrupted" + ], + "description": "interrupted: the job service stopped while the job ran, after it had run it again once from its request, so it has no result; error says so." }, "method": { "type": "string", @@ -7646,7 +7648,14 @@ }, "error": { "type": "string", - "description": "Why a job failed." + "description": "Why a job failed, or why it was interrupted." + }, + "interruptions": { + "type": "array", + "items": { + "type": "integer" + }, + "description": "The job service: each time it stopped while the job ran, in milliseconds since 1970. A job that was running when the service stopped is queued again, once, to run from its stored request." }, "submitted_ms": { "type": "integer", @@ -10388,8 +10397,10 @@ "running", "done", "failed", - "cancelled" - ] + "cancelled", + "interrupted" + ], + "description": "interrupted: the job service stopped while the job ran, after it had run it again once from its request, so it has no result; error says so." }, "method": { "type": "string", @@ -17463,7 +17474,14 @@ }, "error": { "type": "string", - "description": "Why a job failed." + "description": "Why a job failed, or why it was interrupted." + }, + "interruptions": { + "type": "array", + "items": { + "type": "integer" + }, + "description": "The job service: each time it stopped while the job ran, in milliseconds since 1970. A job that was running when the service stopped is queued again, once, to run from its stored request." }, "submitted_ms": { "type": "integer", @@ -21186,7 +21204,7 @@ { "name": "get_quantum_job", "title": "Read a quantum job", - "description": "A job's status, and its result when it is done. The service keeps every job across its restarts for its retention (a day on the public service); a job that was running when it stopped comes back interrupted, with the reason.", + "description": "A job's status, and its result when it is done. The service keeps every job across its restarts for its retention (a day on the public service); a job that was running when it stopped is run again once from its request (its interruptions say when), and stopped again it comes back interrupted, with the reason.", "inputSchema": { "type": "object", "required": [ @@ -21228,8 +21246,10 @@ "running", "done", "failed", - "cancelled" - ] + "cancelled", + "interrupted" + ], + "description": "interrupted: the job service stopped while the job ran, after it had run it again once from its request, so it has no result; error says so." }, "method": { "type": "string", @@ -28303,7 +28323,14 @@ }, "error": { "type": "string", - "description": "Why a job failed." + "description": "Why a job failed, or why it was interrupted." + }, + "interruptions": { + "type": "array", + "items": { + "type": "integer" + }, + "description": "The job service: each time it stopped while the job ran, in milliseconds since 1970. A job that was running when the service stopped is queued again, once, to run from its stored request." }, "submitted_ms": { "type": "integer", @@ -43630,8 +43657,10 @@ "running", "done", "failed", - "cancelled" - ] + "cancelled", + "interrupted" + ], + "description": "interrupted: the job service stopped while the job ran, after it had run it again once from its request, so it has no result; error says so." }, "method": { "type": "string", @@ -50705,7 +50734,14 @@ }, "error": { "type": "string", - "description": "Why a job failed." + "description": "Why a job failed, or why it was interrupted." + }, + "interruptions": { + "type": "array", + "items": { + "type": "integer" + }, + "description": "The job service: each time it stopped while the job ran, in milliseconds since 1970. A job that was running when the service stopped is queued again, once, to run from its stored request." }, "submitted_ms": { "type": "integer", diff --git a/wai/wai-quantum-service/README.md b/wai/wai-quantum-service/README.md index 753cb3a4e..ddf34222f 100644 --- a/wai/wai-quantum-service/README.md +++ b/wai/wai-quantum-service/README.md @@ -179,7 +179,7 @@ Every refusal says why and ends with its code, which the body also carries as `c ## Status and limits -A job is `queued`, `running`, `done`, `failed`, `cancelled`, or `interrupted`: the service stopped while it ran, so it has no result, and `error` says so. Follow a job by its event stream (`GET /v1/jobs/{id}/events`), or by polling `GET /v1/jobs/{id}`, with `?wait=S` to hold each poll until the status changes. +A job is `queued`, `running`, `done`, `failed`, `cancelled`, or `interrupted`: the service stopped while it ran, after it had run it again once, so it has no result, and `error` says so. Follow a job by its event stream (`GET /v1/jobs/{id}/events`), or by polling `GET /v1/jobs/{id}`, with `?wait=S` to hold each poll until the status changes. Every answer carries the client's submission window in the rate-limit fields of the IETF HTTP API working group: `RateLimit-Policy: "submissions";q=30;w=600` (at most 30 submissions in 600 s) and `RateLimit: "submissions";r=12;t=34` (12 left, one more in 34 s). A refusal past the limits carries `Retry-After`. @@ -199,7 +199,8 @@ The MCP tools that run (`run_quantum_circuit`, `run_quantum_set`, `run_on_target Every job is in the service's store before its submission is answered, and again at each change of its status. Programs, requests, results and receipts are kept under the BLAKE3 digests of their bytes; a job's `digests` names them. When the service starts, it reloads the store: - a finished job is served as it was, its receipt still verifying; -- a job that was running comes back `interrupted`, with the reason; +- a job that was running is run again, once, from its stored request: it is queued again in its order, and its `interruptions` say when the service stopped; +- a job the service stops again while it runs once more comes back `interrupted`, with the reason, so a job that stops the service whenever it runs does not run for ever; - a queued one is queued again, in its order. Finished jobs are kept for `WAI_QSVC_RETAIN_SECS` (a day here), then dropped. A record names its client by a keyed hash of the address, never the address. [docs/self-host.md](docs/self-host.md#the-job-store) describes the store. diff --git a/wai/wai-quantum-service/docs/provenance.md b/wai/wai-quantum-service/docs/provenance.md index 5bb433725..905745fc9 100644 --- a/wai/wai-quantum-service/docs/provenance.md +++ b/wai/wai-quantum-service/docs/provenance.md @@ -52,7 +52,7 @@ A job that asks for no compile says so, `"status": "not-applicable"`, with the r |---|---|---| | The job's identifier | `id` (128 random bits) | not sealed; the service's | | Its status and times | `status`, `submitted_ms`, `started_ms`, `finished_ms`; its status as events (`GET /v1/jobs/{id}/events`) | not sealed; the service's | -| Interruption | `status: interrupted` and `error`, for a job the service stopped while it ran | not sealed | +| Interruption | `interruptions`, each time the service stopped while the job ran; a job run again once is sealed as any job is, and `status: interrupted` and `error` for one it stopped again | not sealed | | Shots and seed | `request.shots`, `request.seed`; `result.provenance.execution` | sealed | | The plan it ran under | `estimate` (memory, work) beside the job; `result.provenance.execution.plan` | sealed (in provenance) | | The results | `result.counts`, `expectations`, `amplitudes`, `dynamic`, `ideal` | sealed | diff --git a/wai/wai-quantum-service/docs/self-host.md b/wai/wai-quantum-service/docs/self-host.md index e62a5c7ac..4a82dfc22 100644 --- a/wai/wai-quantum-service/docs/self-host.md +++ b/wai/wai-quantum-service/docs/self-host.md @@ -68,7 +68,7 @@ Every job is written to the store before the request that admitted it is answere ``` - **Writes are whole.** Every file is written under a temporary name, synced, renamed into place, and its directory synced, so a crash leaves the old file or the new one. A second process on the same store is refused. -- **A restart loses nothing it answered.** On start the service reloads the store. A finished job is served as it was, and its receipt still verifies. A job that was running comes back `interrupted`, with the reason; its request is kept, and sent again it runs to the same result. A queued job is queued again, in its order. A worker whose service has gone stops itself. +- **A restart loses nothing it answered.** On start the service reloads the store. A finished job is served as it was, and its receipt still verifies. A job that was running is run again, once, from its stored request: it is queued again in its order, and its `interruptions` say when the service stopped. Stopped again while it runs once more, it comes back `interrupted`, with the reason; its request is kept, and sent again it runs to the same result. A queued job is queued again, in its order. A worker whose service has gone stops itself. - **Objects are checked.** An object is checked against its digest when it is read, and a record that does not read is set aside, not served. - **Retention.** A finished job is kept for `WAI_QSVC_RETAIN_SECS`, then dropped with every object no other job names. - **No addresses.** A record names its client by a keyed hash of the client's address, so it can count the client's limits after a restart and says nothing else. diff --git a/wai/wai-quantum-service/src/discovery.rs b/wai/wai-quantum-service/src/discovery.rs index 165d3295c..637bbf5ca 100644 --- a/wai/wai-quantum-service/src/discovery.rs +++ b/wai/wai-quantum-service/src/discovery.rs @@ -203,7 +203,8 @@ pub fn openapi(server: &str) -> Value { "type": "object", "properties": { "id": { "type": "string" }, - "status": { "type": "string", "enum": ["queued", "running", "done", "failed", "cancelled", "interrupted"], "description": "interrupted: the service stopped while the job ran, so it has no result; error says so, and the request is kept." }, + "status": { "type": "string", "enum": ["queued", "running", "done", "failed", "cancelled", "interrupted"], "description": "interrupted: the service stopped while the job ran, after it had run it again once, so it has no result; error says so, and the request is kept." }, + "interruptions": { "type": "array", "items": { "type": "integer" }, "description": "Each time the service stopped while the job ran, in milliseconds since 1970. A job that was running when the service stopped is queued again, once, in its order, to run from its stored request." }, "method": { "type": "string" }, "why": { "type": "string", "description": "Why this method: what it is, and what the circuit is." }, "request": { "$ref": "#/components/schemas/JobRequest", "description": "The request as submitted: the receipt is checked against it." }, diff --git a/wai/wai-quantum-service/src/http.rs b/wai/wai-quantum-service/src/http.rs index 65165e0eb..72a4ff5fa 100644 --- a/wai/wai-quantum-service/src/http.rs +++ b/wai/wai-quantum-service/src/http.rs @@ -220,7 +220,7 @@ async fn describe(State(app): State) -> Response { "kind": app.service.store().kind(), "durable": app.service.store().durable(), "addresses": "BLAKE3 digests of each program, request, result and receipt", - "restored": { "finished": app.service.restored.finished, "interrupted": app.service.restored.interrupted, "requeued": app.service.restored.requeued, "unreadable": app.service.restored.unreadable }, + "restored": { "finished": app.service.restored.finished, "interrupted": app.service.restored.interrupted, "requeued": app.service.restored.requeued, "rerun": app.service.restored.rerun, "unreadable": app.service.restored.unreadable }, }, "energy": crate::energy::meter().describe(), "receipt_key": app.signer.public_key(), diff --git a/wai/wai-quantum-service/src/jobs.rs b/wai/wai-quantum-service/src/jobs.rs index f62456dc1..13e5d8f6a 100644 --- a/wai/wai-quantum-service/src/jobs.rs +++ b/wai/wai-quantum-service/src/jobs.rs @@ -35,10 +35,14 @@ //! change of its status, before the change is answered: its request and its //! programs when it is admitted, its result and its receipt when it ends, //! each content-addressed. On start the service reloads them: a finished job -//! is served as it was, its receipt still verifying; a job that was running -//! when the service stopped comes back `interrupted`, with the reason; a -//! queued one is queued again, in its order. Finished jobs are kept for -//! [`Config::retain`], then dropped with whatever no other job names. +//! is served as it was, its receipt still verifying; a queued one is +//! queued again, in its order. A job that was running when the service +//! stopped is run again, once ([`RERUNS`]), from its stored request: it is +//! queued again in its order, and its `interruptions` say when the service +//! stopped. Stopped again while it runs, it comes back `interrupted`, with +//! the reason, so a job that stops the service whenever it runs does not +//! run for ever. Finished jobs are kept for [`Config::retain`], then +//! dropped with whatever no other job names. //! - **Retries.** A submission may carry a key of the client's making //! ([`Key`]): sent again with the same key and the same request, it is //! answered with the job the key names, before any limit, and counts as @@ -215,6 +219,9 @@ pub struct Job { /// The digest of the key it was sent with, and its client's name /// ([`Service::key_digest`]), if it was sent with one. pub key: Option, + /// When the service stopped while the job ran, in milliseconds since + /// 1970, each time it did. + pub interruptions: Vec, child: Option, } @@ -248,6 +255,9 @@ impl Job { if self.peak_rss > 0 { o["peak_rss_bytes"] = json!(self.peak_rss); } + if !self.interruptions.is_empty() { + o["interruptions"] = json!(self.interruptions); + } if let Some(r) = &self.result { o["result"] = r.clone(); } @@ -301,6 +311,9 @@ impl Job { if let Some(k) = &self.key { r["key"] = json!(k); } + if !self.interruptions.is_empty() { + r["interruptions"] = json!(self.interruptions); + } if let Work::Set { plans, .. } = &self.work { r["set_plans"] = json!(plans); } @@ -335,6 +348,8 @@ pub struct Restored { pub finished: usize, pub interrupted: usize, pub requeued: usize, + /// Jobs that were running, queued again to run once more. + pub rerun: usize, pub unreadable: usize, } @@ -374,13 +389,20 @@ fn job_id() -> String { b.iter().map(|x| format!("{x:02x}")).collect() } -/// The reason a job that was running when the service stopped carries. -pub fn interrupted_why(at_ms: u64) -> String { +/// How often a job that was running when the service stopped is run again +/// from its stored request: once. +pub const RERUNS: usize = 1; + +/// The reason a job carries that the service stopped while it ran each of +/// the `times` it ran. +pub fn interrupted_why(at_ms: u64, times: usize) -> String { format!( - "interrupted: the service stopped while this job ran (it restarted at {at_ms} ms since the epoch), so its worker did not finish and the job has no result; its request is kept above, and sent again it runs to the same result" + "interrupted: the service stopped while this job ran, each of the {times} times it ran (it restarted last at {at_ms} ms since the epoch); a job is run again once from its stored request and no more, so this one has no result; its request is kept above, and sent again it runs to the same result" ) } +const _: () = assert!(RERUNS == 1, "interrupted_why says once"); + impl Service { /// A service that keeps its jobs in memory alone. #[cfg(test)] @@ -418,9 +440,10 @@ impl Service { } /// A service over `store`, its jobs reloaded: finished ones as they - /// were, running ones `interrupted` with the reason, queued ones queued + /// were; running ones queued again to run once more, or `interrupted` + /// with the reason when they were run again already; queued ones queued /// again in their order (planned again under `config`; one that no longer - /// plans fails, saying why), expired ones dropped. + /// plans fails, saying why); expired ones dropped. pub fn open(config: Config, store: Box) -> Result, String> { let (records, unreadable) = store.records()?; for id in &unreadable { @@ -445,14 +468,24 @@ impl Service { let _ = s.store.remove_record(&id); continue; } - match job.status { - Status::Running => { + // The service stopped while it ran: it runs again from its + // request, queued in its order, unless it ran again already. + let rerun = job.status == Status::Running; + if rerun { + job.interruptions.push(now); + if job.interruptions.len() <= RERUNS { + job.status = Status::Queued; + job.started_ms = None; + job.peak_rss = 0; + } else { job.status = Status::Interrupted; job.finished_ms = Some(now); - job.error = Some(interrupted_why(now)); + job.error = Some(interrupted_why(now, job.interruptions.len())); s.restored.interrupted += 1; s.store.put_record(&job.id, &job.record())?; } + } + match job.status { Status::Queued => { let cap = s.config.job_mib << 20; let replanned = match &job.work { @@ -464,7 +497,11 @@ impl Service { job.plan = p; job.work = w; queued.push((job.submitted_ms, job.id.clone())); - s.restored.requeued += 1; + if rerun { + s.restored.rerun += 1; + } else { + s.restored.requeued += 1; + } } Err(e) => { job.status = Status::Failed; @@ -548,6 +585,7 @@ impl Service { receipt: s("receipt"), }, key: s("key"), + interruptions: rec["interruptions"].as_array().into_iter().flatten().filter_map(Value::as_u64).collect(), child: None, }) } @@ -720,6 +758,7 @@ impl Service { peak_rss: 0, digests, key: key.clone(), + interruptions: Vec::new(), child: None, }; { @@ -1433,6 +1472,19 @@ mod tests { let s = Service::open(cfg(), Box::new(store::Files::open(&dir.join("store")).unwrap())).unwrap(); assert_eq!(s.named("a", Key("q-key"), &req).unwrap(), Some(queued.clone())); assert_eq!(s.submit_work("a", req.clone(), plan(1 << 20), Work::One, Some(Key("q-key"))).unwrap(), queued); + // The running one runs again, under its key. + assert_eq!(s.get(&running).unwrap()["status"], "queued"); + assert_eq!(s.named("a", Key("r-key"), &run).unwrap(), Some(running.clone())); + // Stopped again while it runs, it is interrupted, and its key runs + // the request again. + { + let mut jobs = s.jobs.lock().unwrap(); + let j = jobs.get_mut(&running).unwrap(); + j.status = Status::Running; + s.keep(j).unwrap(); + } + drop(s); + let s = Service::open(cfg(), Box::new(store::Files::open(&dir.join("store")).unwrap())).unwrap(); assert_eq!(s.get(&running).unwrap()["status"], "interrupted"); assert_eq!(s.named("a", Key("r-key"), &run).unwrap(), None); let again = s.submit_work("a", run.clone(), plan(1 << 20), Work::One, Some(Key("r-key"))).unwrap(); @@ -1441,6 +1493,80 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// A job that was running when the service stopped runs again, once, + /// from its stored request: queued in its order, it finishes with a + /// receipt that verifies. Stopped again while it runs, it is + /// interrupted, saying so, and is not run a third time. + #[test] + fn a_job_the_service_stopped_runs_again_once() { + let dir = store_dir("rerun"); + let req = json!({ "qasm": "OPENQASM 3.0; qubit[2] q; bit[2] c; h q[0]; cx q[0], q[1]; c = measure q;", "shots": 16 }); + let p = crate::catalog::plan(&req, 1 << 30).unwrap(); + let key = Arc::new(crate::receipt::Signer::load_or_create(&dir.join("key")).unwrap()); + let open = || { + let s = Service::open(Config { per_client_active: 4, per_client_burst: 10, ..config() }, Box::new(store::Files::open(&dir.join("store")).unwrap())).unwrap(); + s.seal_with(Arc::clone(&key)); + s + }; + // As the scheduler leaves a job it starts. + let start = |s: &Service, id: &str| { + let mut jobs = s.jobs.lock().unwrap(); + let j = jobs.get_mut(id).unwrap(); + j.status = Status::Running; + j.started_ms = Some(now_ms()); + s.keep(j).unwrap(); + }; + let (a, b); + { + let s = open(); + a = s.submit("203.0.113.7", req.clone(), p.clone()).unwrap(); + b = s.submit("203.0.113.7", req.clone(), p.clone()).unwrap(); + start(&s, &a); + start(&s, &b); + } + // Reopened, both are queued again, each with the time the service stopped. + let s = open(); + assert_eq!((s.restored.rerun, s.restored.interrupted, s.restored.requeued), (2, 0, 0), "{:?}", s.restored); + for id in [&a, &b] { + let j = s.get(id).unwrap(); + assert_eq!(j["status"], "queued", "{j}"); + assert_eq!(j["interruptions"].as_array().unwrap().len(), 1, "{j}"); + assert!(j.get("started_ms").is_none() && j.get("error").is_none() && j.get("result").is_none(), "{j}"); + } + assert_eq!(s.queue.lock().unwrap().len(), 2); + // One runs again to its end: its result is sealed and verifies. + start(&s, &a); + let result = wai_quantum::engine::run::run(&req, &p).unwrap(); + s.finish(&a, Ran { outcome: Ok(result), peak: 1, energy: json!({ "unmetered": "a test" }), elapsed_ms: 3 }); + let done = s.get(&a).unwrap(); + assert_eq!(done["status"], "done", "{done}"); + assert!(wai_quantum::engine::job::verify(&json!({ "request": done["request"], "result": done["result"] })).unwrap().0, "{done}"); + // The other is stopped again while it runs. + start(&s, &b); + drop(s); + let s = open(); + assert_eq!((s.restored.rerun, s.restored.interrupted, s.restored.finished), (0, 1, 1), "{:?}", s.restored); + let j = s.get(&b).unwrap(); + assert_eq!(j["status"], "interrupted", "{j}"); + assert_eq!(j["interruptions"].as_array().unwrap().len(), 2, "{j}"); + assert!(j["error"].as_str().unwrap().starts_with("interrupted: the service stopped while this job ran, each of the 2 times it ran"), "{j}"); + assert!(s.queue.lock().unwrap().is_empty(), "it is not run a third time"); + // Both read as the job schema says. + let schemas = crate::mcp::schemas(); + for id in [&a, &b] { + let v = s.get(id).unwrap(); + let errs = schemas.check("service-tools.v1.json#/$defs/job", &v); + assert!(errs.is_empty(), "{}: {v}", errs.join("; ")); + } + // Opened once more, it stays as it is. + drop(s); + let s = open(); + assert_eq!(s.get(&b).unwrap()["interruptions"].as_array().unwrap().len(), 2); + assert_eq!((s.restored.interrupted, s.restored.rerun), (0, 0)); + drop(s); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn a_client_holds_at_most_one_jobs_worth_of_memory_at_once() { let s = Service::new(Config { per_client_active: 5, per_client_burst: 10, ..config() }); @@ -1530,7 +1656,7 @@ mod tests { let key = wai_quantum::ed25519_dalek::SigningKey::from_bytes(&[19; 32]); let request = json!({ "qasm": "OPENQASM 3.0; qubit[2] q; h q[0]; cx q[0], q[1];", "shots": 20, "observables": ["Z0 Z1"] }); let (p, mut result) = wai_quantum::engine::job::plan_and_run(&request, 1 << 30).unwrap(); - let mut job = Job { id: job_id(), client: "test".into(), request: request.clone(), plan: p, work: Work::One, status: Status::Queued, submitted_ms: now_ms(), started_ms: None, finished_ms: None, result: None, error: None, peak_rss: 0, digests: Digests::default(), key: None, child: None }; + let mut job = Job { id: job_id(), client: "test".into(), request: request.clone(), plan: p, work: Work::One, status: Status::Queued, submitted_ms: now_ms(), started_ms: None, finished_ms: None, result: None, error: None, peak_rss: 0, digests: Digests::default(), key: None, interruptions: Vec::new(), child: None }; let r = crate::mcp::schemas(); let check = |j: &Job| { let v = j.to_json(); @@ -1560,7 +1686,7 @@ mod tests { let items = wai_quantum::engine::sets::expand(&set_request).unwrap(); let plans = wai_quantum::engine::sets::plan(&items, 1 << 30).unwrap(); let body = wai_quantum::engine::sets::run(&set_request, 1 << 30, &key, json!({ "unmetered": "a test" })).unwrap(); - let set = Job { id: job_id(), client: "test".into(), request: set_request, plan: wai_quantum::engine::sets::summary(&plans), work: Work::Set { items, plans }, status: Status::Done, submitted_ms: now_ms(), started_ms: Some(now_ms()), finished_ms: Some(now_ms()), result: Some(body["result"].clone()), error: None, peak_rss: 1 << 20, digests: Digests::default(), key: None, child: None }; + let set = Job { id: job_id(), client: "test".into(), request: set_request, plan: wai_quantum::engine::sets::summary(&plans), work: Work::Set { items, plans }, status: Status::Done, submitted_ms: now_ms(), started_ms: Some(now_ms()), finished_ms: Some(now_ms()), result: Some(body["result"].clone()), error: None, peak_rss: 1 << 20, digests: Digests::default(), key: None, interruptions: Vec::new(), child: None }; let done = check(&set); assert_eq!(wai_quantum::engine::sets::verify(&done)["valid"], true, "{done}"); } @@ -1634,23 +1760,30 @@ mod tests { assert!(!raw.contains("203.0.113.7") && !raw.contains("198.51.100.1"), "a record keeps no address"); let before = std::fs::read_dir(dir.join("store/objects")).unwrap().flatten().flat_map(|d| std::fs::read_dir(d.path()).unwrap().flatten()).count(); let s = Service::open(Config { per_client_active: 10, per_client_burst: 10, ..config() }, Box::new(store::Files::open(&dir.join("store")).unwrap())).unwrap(); - assert_eq!((s.restored.finished, s.restored.interrupted, s.restored.requeued), (1, 1, 1), "{:?}", s.restored); + assert_eq!((s.restored.finished, s.restored.interrupted, s.restored.requeued, s.restored.rerun), (1, 0, 1, 1), "{:?}", s.restored); let d = s.get(&done).unwrap(); assert_eq!(d["status"], "done"); let (valid, _) = wai_quantum::engine::job::verify(&json!({ "request": d["request"], "result": d["result"] })).unwrap(); assert!(valid, "a reloaded result's receipt still verifies: {d}"); assert_eq!(d["digests"]["request"].as_str().unwrap(), store::digest(&canon::canonical(&req))); + // The running one is queued again to run once more, saying when the + // service stopped. let r = s.get(&running).unwrap(); - assert_eq!(r["status"], "interrupted"); - assert!(r["error"].as_str().unwrap().starts_with("interrupted: the service stopped while this job ran"), "{r}"); - assert!(r.get("result").is_none()); + assert_eq!(r["status"], "queued", "{r}"); + assert_eq!(r["interruptions"].as_array().unwrap().len(), 1, "{r}"); + assert!(r.get("result").is_none() && r.get("error").is_none() && r.get("started_ms").is_none(), "{r}"); assert_eq!(s.get(&queued).unwrap()["status"], "queued"); - assert_eq!(s.queue.lock().unwrap().iter().cloned().collect::>(), vec![queued.clone()]); + let mut q: Vec = s.queue.lock().unwrap().iter().cloned().collect(); + q.sort(); + let mut want = vec![running.clone(), queued.clone()]; + want.sort(); + assert_eq!(q, want); assert!(s.get(&old).is_none(), "an expired job is not reloaded"); let after = std::fs::read_dir(dir.join("store/objects")).unwrap().flatten().flat_map(|d| std::fs::read_dir(d.path()).unwrap().flatten()).count(); assert!(after < before, "the expired job's own objects are swept ({before} → {after})"); - // The requeued job still counts against its client. + // The requeued jobs still count against their clients. assert_eq!(s.holding("198.51.100.1").0, 1); + assert_eq!(s.holding("203.0.113.7").0, 1); drop(s); let _ = std::fs::remove_dir_all(&dir); } @@ -1667,7 +1800,7 @@ mod tests { assert!(t["seconds"].is_null() && t["at_most"] == 5, "{t}"); // Timed jobs here take over, and their metered power gives an energy estimate. for _ in 0..3 { - let job = Job { id: job_id(), client: "x".into(), request: json!({}), plan: Plan { work: 1_000_000_000, ..p.clone() }, work: Work::One, status: Status::Done, submitted_ms: 0, started_ms: None, finished_ms: Some(1), result: Some(json!({ "elapsed_ms": 500, "energy": { "joules_micro": 5_000_000, "energy_uncertainty": { "window_us": 500_000 } } })), error: None, peak_rss: 0, digests: Digests::default(), key: None, child: None }; + let job = Job { id: job_id(), client: "x".into(), request: json!({}), plan: Plan { work: 1_000_000_000, ..p.clone() }, work: Work::One, status: Status::Done, submitted_ms: 0, started_ms: None, finished_ms: Some(1), result: Some(json!({ "elapsed_ms": 500, "energy": { "joules_micro": 5_000_000, "energy_uncertainty": { "window_us": 500_000 } } })), error: None, peak_rss: 0, digests: Digests::default(), key: None, interruptions: Vec::new(), child: None }; s.time(&job); } let (t, e) = s.estimate(&p); diff --git a/wai/wai-quantum-service/src/main.rs b/wai/wai-quantum-service/src/main.rs index 7b9a89313..ce3b2ff7b 100644 --- a/wai/wai-quantum-service/src/main.rs +++ b/wai/wai-quantum-service/src/main.rs @@ -193,12 +193,13 @@ fn serve() { let console = std::env::var("WAI_QSVC_CONSOLE").unwrap_or_else(|_| "https://wai.transaction.science/quantum-run/".into()); let r = service.restored; eprintln!( - "job store: {} ({}); reloaded {} finished, {} interrupted, {} queued again, {} unreadable", + "job store: {} ({}); reloaded {} finished, {} interrupted, {} queued again, {} run again after an interruption, {} unreadable", service.store().kind(), if service.store().durable() { "kept across restarts" } else { "in memory only" }, r.finished, r.interrupted, r.requeued, + r.rerun, r.unreadable ); eprintln!("energy: {}", meter.describe()); diff --git a/wai/wai-quantum-service/src/mcp.rs b/wai/wai-quantum-service/src/mcp.rs index 16c93e0dc..29c59d633 100644 --- a/wai/wai-quantum-service/src/mcp.rs +++ b/wai/wai-quantum-service/src/mcp.rs @@ -278,7 +278,7 @@ fn tool_list() -> Value { { "name": "get_quantum_job", "title": "Read a quantum job", - "description": "A job's status, and its result when it is done. The service keeps every job across its restarts for its retention (a day on the public service); a job that was running when it stopped comes back interrupted, with the reason.", + "description": "A job's status, and its result when it is done. The service keeps every job across its restarts for its retention (a day on the public service); a job that was running when it stopped is run again once from its request (its interruptions say when), and stopped again it comes back interrupted, with the reason.", "inputSchema": { "type": "object", "required": ["id"], "properties": { "id": { "type": "string" } } }, "annotations": { "readOnlyHint": true } }, diff --git a/wai/wai-quantum-service/src/store.rs b/wai/wai-quantum-service/src/store.rs index e8457ac30..2c75c5634 100644 --- a/wai/wai-quantum-service/src/store.rs +++ b/wai/wai-quantum-service/src/store.rs @@ -413,8 +413,10 @@ mod tests { assert_eq!(fs::metadata(root.join("client.key")).unwrap().permissions().mode() & 0o777, 0o600); } let p = pseudonym(&a, "203.0.113.7"); - assert_eq!(p.len(), 32); - assert!(!p.contains("203")); + // Hex of a keyed hash: it cannot spell the address, whose dots hex has + // none of. (Three digits of the address can turn up in it by chance, + // as they did about once in 140 runs when this held them out.) + assert!(p.len() == 32 && p.bytes().all(|b| b.is_ascii_hexdigit()), "{p}"); assert_eq!(p, pseudonym(&a, "203.0.113.7")); assert_ne!(p, pseudonym(&[0; 32], "203.0.113.7")); let _ = fs::remove_dir_all(&root); diff --git a/wai/wai-quantum-service/tests/restart.rs b/wai/wai-quantum-service/tests/restart.rs index a0d9cd09e..5a66e4951 100644 --- a/wai/wai-quantum-service/tests/restart.rs +++ b/wai/wai-quantum-service/tests/restart.rs @@ -1,7 +1,8 @@ //! The service killed mid-job and started again on the same store: every //! result it gave is served again with its receipt, which still verifies; -//! the job it was running comes back interrupted, with the reason; the one -//! queued behind it is queued again. +//! the job it was running runs again, once; the one queued behind it is +//! queued again. Killed while that job runs again, the job comes back +//! interrupted, with the reason, and the queue goes on. use serde_json::{Value, json}; use std::io::{Read, Write}; @@ -114,17 +115,38 @@ fn a_service_killed_mid_job_serves_its_results_again_and_says_what_it_interrupte assert_eq!(wai_quantum::engine::sets::verify(&set_again)["valid"], true, "{set_again}"); assert_eq!(set_again["result"]["items"][1]["point"], json!({ "theta": 0.7 })); - // The job it was running: interrupted, with the reason, its request kept. + // The description says what was found: the running job run again, the + // queued one queued again. + let restored = |port: u16| { + let (_, d) = json_of(port, "GET", "/v1", None); + assert_eq!(d["store"]["durable"], true, "{}", d["store"]); + let r = &d["store"]["restored"]; + (r["finished"].as_u64(), r["interrupted"].as_u64(), r["requeued"].as_u64(), r["rerun"].as_u64()) + }; + assert_eq!(restored(port), (Some(2), Some(0), Some(1), Some(1))); + + // The job it was running runs again from its request, first in its + // order, saying when the service stopped. + let r = until(port, &running, "running"); + assert_eq!(r["interruptions"].as_array().map(Vec::len), Some(1), "{r}"); + assert!(r.get("result").is_none() && r.get("error").is_none() && r["request"] == long, "{r}"); + assert_eq!(json_of(port, "GET", &format!("/v1/jobs/{queued}"), None).1["status"], "queued"); + + // Killed again while it runs again: interrupted, with the reason, its + // request kept, and not run a third time. + svc.kill().unwrap(); + svc.wait().unwrap(); + let mut svc = start(&dir, port); + assert_eq!(restored(port), (Some(2), Some(1), Some(1), Some(0))); let (_, r) = json_of(port, "GET", &format!("/v1/jobs/{running}"), None); assert_eq!(r["status"], "interrupted", "{r}"); - assert!(r["error"].as_str().unwrap().contains("the service stopped while this job ran"), "{r}"); + assert!(r["error"].as_str().unwrap().contains("the service stopped while this job ran, each of the 2 times it ran"), "{r}"); + assert_eq!(r["interruptions"].as_array().map(Vec::len), Some(2), "{r}"); assert!(r.get("result").is_none() && r["request"] == long); // The one queued behind it runs now. let q = until(port, &queued, "done"); assert!(wai_quantum::engine::job::verify(&json!({ "request": q["request"], "result": q["result"] })).unwrap().0); - // And the description says what was found. - let (_, d) = json_of(port, "GET", "/v1", None); - assert_eq!((d["store"]["durable"].as_bool(), d["store"]["restored"]["interrupted"].as_u64(), d["store"]["restored"]["requeued"].as_u64(), d["store"]["restored"]["finished"].as_u64()), (Some(true), Some(1), Some(1), Some(2)), "{}", d["store"]); + assert_eq!(json_of(port, "GET", &format!("/v1/jobs/{running}"), None).1["status"], "interrupted"); svc.kill().unwrap(); svc.wait().unwrap(); diff --git a/wai/wai-quantum/schemas/job.v1.json b/wai/wai-quantum/schemas/job.v1.json index 3b836416a..2acabe306 100644 --- a/wai/wai-quantum/schemas/job.v1.json +++ b/wai/wai-quantum/schemas/job.v1.json @@ -7,7 +7,7 @@ "required": ["status", "request"], "properties": { "id": { "type": "string", "description": "The job service's id for the job." }, - "status": { "enum": ["queued", "running", "done", "failed", "cancelled"] }, + "status": { "enum": ["queued", "running", "done", "failed", "cancelled", "interrupted"], "description": "interrupted: the job service stopped while the job ran, after it had run it again once from its request, so it has no result; error says so." }, "method": { "type": "string", "description": "The method the plan chose; set for a set." }, "why": { "type": "string", "description": "Why the plan chose the method." }, "estimate": { @@ -21,7 +21,8 @@ }, "request": { "$ref": "job-request.v1.json" }, "result": { "anyOf": [{ "$ref": "job-result.v1.json" }, { "$ref": "set.v1.json" }] }, - "error": { "type": "string", "description": "Why a job failed." }, + "error": { "type": "string", "description": "Why a job failed, or why it was interrupted." }, + "interruptions": { "type": "array", "items": { "type": "integer" }, "description": "The job service: each time it stopped while the job ran, in milliseconds since 1970. A job that was running when the service stopped is queued again, once, to run from its stored request." }, "submitted_ms": { "type": "integer", "description": "The job service: when it was submitted, in milliseconds since 1970." }, "started_ms": { "type": "integer" }, "finished_ms": { "type": "integer" }, diff --git a/wai/wai-quantum/src/engine/schema.rs b/wai/wai-quantum/src/engine/schema.rs index 0e07cd321..151578fe0 100644 --- a/wai/wai-quantum/src/engine/schema.rs +++ b/wai/wai-quantum/src/engine/schema.rs @@ -795,6 +795,8 @@ mod tests { // The service's form: an id, times, and a job not yet done. ok(r.check("job.v1.json", &json!({ "id": "a1", "status": "running", "method": "stabilizer", "why": "Clifford", "estimate": { "memory_bytes": 1, "work": 1 }, "request": { "qasm": bell }, "submitted_ms": 1, "started_ms": 2 })), "a running job"); ok(r.check("job.v1.json", &json!({ "id": "a1", "status": "failed", "method": "statevector", "why": "", "request": { "qasm": bell }, "error": "the job passed its memory", "submitted_ms": 1, "finished_ms": 3, "peak_rss_bytes": 9 })), "a failed job"); + ok(r.check("job.v1.json", &json!({ "id": "a1", "status": "queued", "method": "statevector", "why": "", "request": { "qasm": bell }, "submitted_ms": 1, "interruptions": [5] })), "a job queued again to run once more"); + ok(r.check("job.v1.json", &json!({ "id": "a1", "status": "interrupted", "method": "statevector", "why": "", "request": { "qasm": bell }, "error": "interrupted: the service stopped while this job ran", "submitted_ms": 1, "started_ms": 6, "finished_ms": 9, "interruptions": [5, 9] })), "an interrupted job"); } /// Every error-correction answer (a code with its certificate and a given -- 2.54.0 (Apple Git-157)