Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

14 changes: 7 additions & 7 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ members = [
resolver = "2"

[workspace.package]
version = "3.52.3"
version = "3.52.8"
edition = "2024"
license-file = "LICENSE"
authors = ["PureTensor, Inc. <[email protected]>"]
Expand Down Expand Up @@ -52,12 +52,12 @@ rmcp = { version = "3.3.0", features = ["server", "macros", "schemars", "transpo
futures-util = { version = "0.3", default-features = false }

# Internal crates
ptask-core = { path = "crates/ptask-core", version = "3.52.3" }
ptask-tui = { path = "crates/ptask-tui", version = "3.52.3" }
ptask-server = { path = "crates/ptask-server", version = "3.52.3" }
ptask-bot = { path = "crates/ptask-bot", version = "3.52.3" }
ptask-distill = { path = "crates/ptask-distill", version = "3.52.3" }
ptask-notify = { path = "crates/ptask-notify", version = "3.52.3" }
ptask-core = { path = "crates/ptask-core", version = "3.52.8" }
ptask-tui = { path = "crates/ptask-tui", version = "3.52.8" }
ptask-server = { path = "crates/ptask-server", version = "3.52.8" }
ptask-bot = { path = "crates/ptask-bot", version = "3.52.8" }
ptask-distill = { path = "crates/ptask-distill", version = "3.52.8" }
ptask-notify = { path = "crates/ptask-notify", version = "3.52.8" }

# HMAC for outbound webhook signing (v0.3.5).
sha2 = "0.10"
Expand Down
100 changes: 96 additions & 4 deletions crates/ptask-core/src/approval_journal.rs
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,8 @@ pub struct Journal {
key_file: PathBuf,
}

/// One journal line. `ev` is request, approve, reject, withdraw or consume.
/// One journal line. `ev` is request, approve, reject, withdraw, consume,
/// or consume_void (a consume whose database transaction did not commit).
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct Record {
pub n: u64,
Expand Down Expand Up @@ -251,23 +252,97 @@ impl Journal {
/// consume record under the same lock so a replay is refused across
/// processes. Shadow mode logs a refusal and still records the consume
/// when the chain is intact.
pub fn check_and_record_consume(&self, claim: &Claim, actor: &str, ts: &str) -> Result<()> {
///
/// `Ok(true)` means a consume line was appended. `Ok(false)` is shadow
/// mode softening a failure that wrote nothing. Callers that void a
/// consume when their own database transaction rolls back must void
/// only after `Ok(true)`: a refusal leaves any earlier committed
/// consume in place. A shadow refusal appends a consume only when none
/// is open; an already-open consume returns `Ok(false)` and writes
/// nothing, so a rolled-back retry cannot void it.
pub fn check_and_record_consume(&self, claim: &Claim, actor: &str, ts: &str) -> Result<bool> {
let res = self.locked(|key, file, chain| {
let verdict = authorise(&chain, claim);
match verdict {
Ok(()) => {
Self::append_in(key, file, &chain, Self::record("consume", claim, actor, ts))
Self::append_in(key, file, &chain, Self::record("consume", claim, actor, ts))?;
Ok(true)
}
Err(e) if self.mode == Mode::Shadow => {
tracing::warn!(
approval = claim.uuid,
"approval journal (shadow) would refuse consume: {e}"
);
Self::append_in(key, file, &chain, Self::record("consume", claim, actor, ts))
// An open consume already committed. Another line and
// Ok(true) would arm ConsumeVoidGuard; a rolled-back
// retry would then void it and reopen the approval.
if consume_is_open(&chain, claim.uuid) {
Ok(false)
} else {
Self::append_in(
key,
file,
&chain,
Self::record("consume", claim, actor, ts),
)?;
Ok(true)
}
}
Err(e) => Err(e),
}
});
match res {
Ok(appended) => Ok(appended),
Err(e) if self.mode == Mode::Shadow => {
tracing::warn!("approval journal (shadow) error ignored: {e}");
Ok(false)
}
Err(e) => Err(e),
}
}

/// Whether the journal currently authorises `claim` (an approve is
/// present, unconsumed, and matches). Does not append. Shadow mode
/// logs a refusal and returns `Ok`. Used by `payload` and `verify`,
/// which release or attest before `consume` latches.
pub fn authorises(&self, claim: &Claim) -> Result<()> {
let res = self.locked(|_key, _file, chain| authorise(&chain, claim));
match res {
Err(e) if self.mode == Mode::Shadow => {
tracing::warn!(
approval = claim.uuid,
"approval journal (shadow) would refuse release: {e}"
);
Ok(())
}
other => other,
}
}

/// Append `consume_void` when `uuid` has an unvoided consume.
///
/// `consume` writes the journal line before the SQLite commit, because
/// the journal is the authority and must not lag a committed latch.
/// If that transaction then rolls back, the consume line would brick
/// the still-approved row forever (`authorise` treats consume as
/// sticky). Only a holder of the journal key can append the void, and
/// the void does not look at the database: clearing `consumed_at` in
/// SQLite must not undo a consume that did commit.
///
/// No-op when this uuid has no open consume (the check failed before
/// any line was written, or a void is already in place).
pub fn void_uncommitted_consume(&self, claim: &Claim, actor: &str, ts: &str) -> Result<()> {
let res = self.locked(|key, file, chain| {
if !consume_is_open(&chain, claim.uuid) {
return Ok(());
}
Self::append_in(
key,
file,
&chain,
Self::record("consume_void", claim, actor, ts),
)
});
self.soften(res)
}

Expand Down Expand Up @@ -301,6 +376,10 @@ fn authorise(chain: &[Record], claim: &Claim) -> Result<()> {
approve.insert("a", r);
}
"consume" => consumed = true,
// A consume whose database transaction rolled back. Only the
// key holder can write this; it does not clear a consume that
// has no matching void.
"consume_void" => consumed = false,
"reject" | "withdraw" => {
approve.remove("a");
}
Expand Down Expand Up @@ -332,3 +411,16 @@ fn authorise(chain: &[Record], claim: &Claim) -> Result<()> {
}
Ok(())
}

/// True when the latest consume for `uuid` has not been voided.
fn consume_is_open(chain: &[Record], uuid: &str) -> bool {
let mut consumed = false;
for r in chain.iter().filter(|r| r.uuid == uuid) {
match r.ev.as_str() {
"consume" => consumed = true,
"consume_void" => consumed = false,
_ => {}
}
}
consumed
}
81 changes: 78 additions & 3 deletions crates/ptask-core/src/approvals.rs
Original file line number Diff line number Diff line change
Expand Up @@ -910,6 +910,13 @@ pub fn payload_bytes(db: &Db, id: &str) -> Result<Vec<u8>> {
if ap.consumed_at.is_some() {
return Err(Error::Approval(ApprovalError::AlreadyConsumed));
}
// Enforce mode: the executor pattern is payload, then act, then
// consume. A row written straight into SQLite passes the database
// gate and would otherwise release its bytes before consume refused
// it. The journal check does not latch.
if let Some(j) = Journal::from_env() {
j.authorises(&claim_of(&ap))?;
}
stored_payload(ap)
}

Expand Down Expand Up @@ -1363,6 +1370,11 @@ pub fn verify(db: &Db, id: &str, offered: &PayloadSource) -> Result<Approval> {
let ap = get(db, id)?;
let digest = offered_digest(offered)?;
gate_status_and_digest(&ap, &digest, &dates::now_in_operator_tz()?)?;
// Same journal authority as consume, without the latch. A forged
// approved row must not verify in enforce mode.
if let Some(j) = Journal::from_env() {
j.authorises(&claim_of(&ap))?;
}
Ok(ap)
}

Expand Down Expand Up @@ -1390,9 +1402,32 @@ pub fn consume(db: &Db, id: &str, offered: &PayloadSource, ctx: &EventCtx) -> Re
// The journal, not the row, is the authority in enforce mode: a row
// written straight into SQLite has no journaled approval and is refused
// here, inside the transaction, so the latch rolls back.
if let Some(j) = Journal::from_env() {
j.check_and_record_consume(&claim_of(&current), actor, &now)?;
}
//
// The consume line is fsynced before commit so a committed latch can
// never race ahead of the journal. The guard is declared after `tx`
// and dropped first, still holding the write lock: if `record_event`
// or `commit` fails, it appends `consume_void` before the rollback is
// visible. Without that, the journal consume sticks, the row stays
// approved and unconsumed, and every retry is AlreadyConsumed.
let mut void_guard = ConsumeVoidGuard {
journal: Journal::from_env(),
uuid: current.uuid.clone(),
digest: current.digest.clone(),
kind: current.kind.clone(),
expires_at: current.expires_at.clone(),
actor: actor.to_string(),
ts: now.clone(),
armed: false,
};
// Arm only when this call appended a consume. A refusal (forged row,
// or a committed consume whose database latch was cleared) must not
// void an earlier consume that did commit.
let appended = if let Some(j) = void_guard.journal.as_ref() {
j.check_and_record_consume(&claim_of(&current), actor, &now)?
} else {
false
};
void_guard.armed = appended;
record_event(
&tx,
ctx,
Expand All @@ -1401,9 +1436,49 @@ pub fn consume(db: &Db, id: &str, offered: &PayloadSource, ctx: &EventCtx) -> Re
serde_json::json!({"approval_id": current.ap_id()}),
)?;
tx.commit()?;
void_guard.armed = false;
get(db, &current.uuid)
}

/// Voids a journal consume when dropped while still armed.
///
/// Declared after the SQLite transaction so Drop runs while that
/// transaction, and its write lock, is still held.
struct ConsumeVoidGuard {
journal: Option<Journal>,
uuid: String,
digest: String,
kind: String,
expires_at: Option<String>,
actor: String,
ts: String,
armed: bool,
}

impl Drop for ConsumeVoidGuard {
fn drop(&mut self) {
if !self.armed {
return;
}
let Some(journal) = self.journal.as_ref() else {
return;
};
let claim = Claim {
uuid: &self.uuid,
digest: &self.digest,
kind: &self.kind,
expires_at: self.expires_at.as_deref(),
decided_by: None,
};
if let Err(e) = journal.void_uncommitted_consume(&claim, &self.actor, &self.ts) {
tracing::error!(
approval = %self.uuid,
"approval journal consume survived a rolled-back transaction and could not be voided: {e}"
);
}
}
}

/// Mark pending rows whose `expires_at` is in the past as expired.
/// Idempotent: already-expired rows are left alone. Returns how many
/// newly expired. Runs on every request, so it reads only the three
Expand Down
Loading
Loading