Build a Rust resource service
Accept a report only from its current assigned caller, and retain the original mutation outcome in the same SQLite transaction.
On this page
A report writer can lose its assignment while it is still running. This Rust resource refuses that writer's next write even after it rereads the newest version, and still returns the result it committed before the reassignment, without writing again.
One resource-owned SQLite transaction holds the assignment, the document version, operation deduplication, and mutation history. The same library runs as a native tool service in the embedded host project. SQLite ships as a bundled library, so no database service is needed.
Start1. Run the stale-writer exercise
Download the source projects and build the Rust workspace. The exercise reassigns one document from worker A to worker B, then lets A read the newest version and attempt a fresh write.
cargo run --locked --bin chio-resource-example -- report-resource.sqlite3{
"mutations": 2,
"stale_writer": "superseded",
"original_outcome_replayed": true,
"survives_reopen": true
}You should see "stale_writer": "superseded" and exactly two mutations. The executable assigns generation one to A, and A commits its report. The operator transfers ownership to B. A reads the latest version and the new generation and writes again; the transaction compares the authenticated caller with the current owner and refuses it. B publishes once, and A retrieves its original committed outcome without changing the document.
2. Take the caller from the trusted adapter
The library has no dependency on the worker client. Its caller parameter comes from the trusted resource adapter, outside the model-controlled request; the native adapter reads it from ToolInvocationContext::capability_hash(). The worker request has no caller field, and deserialization refuses unknown fields.
Complete Resource implementation: assignment, publication, and replay
//! Resource-owned transaction. The trusted adapter supplies `caller` separately
//! from model-controlled arguments. No mailbox claim bypasses this check.
use anyhow::{ensure, Result};
use rusqlite::{params, Connection, OpenFlags, OptionalExtension, TransactionBehavior};
use serde::{Deserialize, Serialize};
use std::path::Path;
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Publish {
pub operation: String,
pub generation: i64,
pub version: i64,
pub body: String,
}
#[derive(Debug, Serialize, Deserialize, PartialEq)]
pub struct Outcome {
pub status: String,
pub version: i64,
}
pub struct Resource {
db: Connection,
}
impl Resource {
pub fn initialize(path: &Path) -> Result<()> {
std::fs::OpenOptions::new()
.create_new(true)
.write(true)
.open(path)?;
let store = Self::open(path)?;
store.db.execute_batch("PRAGMA journal_mode=WAL;
CREATE TABLE document(id INTEGER PRIMARY KEY CHECK(id=1),version INTEGER NOT NULL,body TEXT NOT NULL,owner TEXT NOT NULL,generation INTEGER NOT NULL);
CREATE TABLE operations(caller TEXT NOT NULL,id TEXT NOT NULL,request TEXT NOT NULL,outcome TEXT NOT NULL,PRIMARY KEY(caller,id));
CREATE TABLE mutations(version INTEGER PRIMARY KEY,caller TEXT NOT NULL,operation TEXT NOT NULL);
INSERT INTO document VALUES(1,0,'draft','unassigned',0);")?;
Ok(())
}
pub fn open(path: &Path) -> Result<Self> {
let db = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_WRITE)?;
db.busy_timeout(std::time::Duration::from_secs(5))?;
db.execute_batch("PRAGMA synchronous=FULL")?;
Ok(Self { db })
}
/// Operator-only assignment; do not expose as a worker tool.
pub fn assign(&mut self, expected_generation: i64, owner: &str) -> Result<i64> {
ensure!(!owner.is_empty() && owner.len() <= 512);
let tx = self
.db
.transaction_with_behavior(TransactionBehavior::Immediate)?;
ensure!(
tx.execute(
"UPDATE document SET owner=?1,generation=generation+1 WHERE id=1 AND generation=?2",
params![owner, expected_generation]
)? == 1,
"assignment conflict"
);
tx.commit()?;
Ok(expected_generation + 1)
}
pub fn read(&self) -> Result<(i64, String, i64)> {
Ok(self.db.query_row(
"SELECT version,body,generation FROM document WHERE id=1",
[],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
)?)
}
pub fn publish(&mut self, caller: &str, request: &Publish) -> Result<Outcome> {
ensure!(
!caller.is_empty()
&& caller.len() <= 512
&& !request.operation.is_empty()
&& request.operation.len() <= 512
);
ensure!(request.body.len() <= 65536 && request.generation >= 0 && request.version >= 0);
let encoded = serde_json::to_string(request)?;
let tx = self
.db
.transaction_with_behavior(TransactionBehavior::Immediate)?;
if let Some((saved, result)) = tx
.query_row(
"SELECT request,outcome FROM operations WHERE caller=?1 AND id=?2",
params![caller, request.operation],
|r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)),
)
.optional()?
{
ensure!(saved == encoded, "operation key reused with changed input");
return Ok(serde_json::from_str(&result)?);
}
let (owner, generation, version): (String, i64, i64) = tx.query_row(
"SELECT owner,generation,version FROM document WHERE id=1",
[],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
)?;
let status = if owner != caller || generation != request.generation {
"superseded"
} else if version != request.version {
"version_conflict"
} else {
"published"
};
let result = Outcome {
status: status.into(),
version: if status == "published" {
version
.checked_add(1)
.ok_or_else(|| anyhow::anyhow!("version overflow"))?
} else {
version
},
};
if status == "published" {
tx.execute(
"UPDATE document SET body=?1,version=?2 WHERE id=1",
params![request.body, result.version],
)?;
tx.execute(
"INSERT INTO mutations VALUES(?1,?2,?3)",
params![result.version, caller, request.operation],
)?;
}
tx.execute(
"INSERT INTO operations VALUES(?1,?2,?3,?4)",
params![
caller,
request.operation,
encoded,
serde_json::to_string(&result)?
],
)?;
tx.commit()?;
Ok(result)
}
pub fn mutations(&self) -> Result<i64> {
Ok(self
.db
.query_row("SELECT COUNT(*) FROM mutations", [], |r| r.get(0))?)
}
}The model can shape the request's arguments, and the adapter alone supplies who is calling. The ownership check in step 3 compares that adapter-supplied caller with the current owner.
3. Check ownership before version, inside one transaction
Commit the mutation and its result together
The caller’s identity comes from the adapter, apart from the request fields the model controls. One immediate transaction looks up the exact request first. A new mutation checks the current owner and assignment generation, then the document version, and commits the document with its outcome so both survive a reopen.
Drawn fromlanguages/rust/resource/src/lib.rs
- Start an immediate transaction. Assignment changes and publication cannot pass each other between the ownership check and the mutation.
- Look up the caller plus operation key. An identical completed request returns its saved result; different arguments under that key are a conflict.
- Check the current owner and generation. A superseded caller stays superseded even with fresh document data.
- Check the expected version. A current owner cannot silently overwrite an intervening document change.
- Commit the mutation and its outcome together. A lost response is recovered from the resource without another mutation.
The replay lookup comes before the assignment check because it returns a result that was already committed. It gives the old worker no new write: any new operation from that caller still meets the assignment check. Whether your data-access policy lets a reassigned caller read results it committed before is your application's decision.
4. Connect the library to the kernel
In production, the tool transport authenticates the kernel-selected caller before passing it into this transaction; a capability hash inside request JSON authenticates nothing. The embedded host shows the native case. Only the kernel constructs the invocation context. The adapter refuses context-free calls, and it serves the context-bearing cost path as well as the ordinary invocation path.
Assignment is an operator operation and is never registered as a worker tool. An assignment API needs its own authenticated authority and an expected-generation condition. A mailbox claim is a delivery lease and leaves the resource's owner as it is.
Read more in Embedded process host.
5. Verify the outcome in the database
The executable closes and reopens the database before it checks the final state, so its assertions read stored rows. Read them yourself with any SQLite client.
SELECT version, body, owner, generation FROM document;
SELECT version, caller, operation FROM mutations ORDER BY version;
SELECT caller, id, outcome FROM operations ORDER BY caller, id;You should see B's report at version two and exactly two mutation rows. The refused stale operation appears in operations with its outcome and has no mutation row. The kernel's receipt for each call records its decision; the operations table records what the resource committed.
A lost response is recovered by its original operation key
If the resource commits and the process host loses its response, the database and the kernel receipt can disagree about what is known. Recover the original process key first. The resource's outcome lookup can show that a mutation committed, and only the kernel signs a successful receipt, so keep the two records distinct when you reconcile an unknown external effect.
A restarted service opens its existing state. Initialization uses exclusive creation and refuses a path that already exists. A busy timeout bounds SQLite contention, and a busy error never justifies discarding the database or retrying under a new operation key.
The one-document model keeps the ownership contract visible. For tenants or several documents, put that identity into every assignment, operation, and mutation key.