feat: add authorized Plugin Ticket data capabilities
This commit is contained in:
@@ -517,8 +517,15 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip
|
||||
"Subject Ticket responsibility and current-body notification routing"
|
||||
.into()
|
||||
),
|
||||
(109, "Durable Plugin Host invocations and HTTPS effect receipts".into()),
|
||||
(110, "Worker Plugin launch selection associations".into())
|
||||
(
|
||||
109,
|
||||
"Durable Plugin Host invocations and HTTPS effect receipts".into()
|
||||
),
|
||||
(110, "Worker Plugin launch selection associations".into()),
|
||||
(
|
||||
111,
|
||||
"Typed Plugin Ticket extension data and atomic mutation receipts".into()
|
||||
)
|
||||
]
|
||||
);
|
||||
for (table, rows) in retained.iter().zip(original.iter()) {
|
||||
|
||||
@@ -14,6 +14,7 @@ pub mod memory_staging;
|
||||
pub mod observation;
|
||||
pub mod plugin_binding;
|
||||
pub mod plugin_catalog;
|
||||
pub mod plugin_data;
|
||||
pub mod plugin_delivery;
|
||||
pub mod plugin_host;
|
||||
pub mod plugin_https;
|
||||
|
||||
@@ -88,8 +88,19 @@ pub(crate) fn migrate(conn: &Connection) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
pub(crate) fn verify(conn: &Connection) -> Result<()> {
|
||||
verify_catalog_schema(conn, false)
|
||||
}
|
||||
/// Latest-schema verification after the separately registered typed-data
|
||||
/// migration. Historical migration preconditions continue to use `verify`.
|
||||
pub(crate) fn verify_with_data(conn: &Connection) -> Result<()> {
|
||||
verify_catalog_schema(conn, true)
|
||||
}
|
||||
fn verify_catalog_schema(conn: &Connection, with_data: bool) -> Result<()> {
|
||||
let expected = Connection::open_in_memory()?;
|
||||
expected.execute_batch(include_str!("plugin_catalog_schema.sql"))?;
|
||||
if with_data {
|
||||
expected.execute_batch(crate::plugin_data::REFERENCE_MIGRATION_SQL)?;
|
||||
}
|
||||
let objects = expected.prepare("SELECT name,sql FROM sqlite_schema WHERE sql IS NOT NULL AND tbl_name LIKE 'plugin_%' ORDER BY name")?
|
||||
.query_map([], |r| Ok((r.get::<_,String>(0)?, r.get::<_,String>(1)?)))?
|
||||
.collect::<rusqlite::Result<Vec<_>>>()?;
|
||||
@@ -249,6 +260,7 @@ fn artifact_metadata(
|
||||
.collect(),
|
||||
connections: decode(&encode(&m.connections)?)?,
|
||||
requested_grants: decode(&encode(&m.requested_grants)?)?,
|
||||
data_collections: decode(&encode(&m.data_collections)?)?,
|
||||
config_schema: PluginSchemaContribution {
|
||||
identity: schema.identity,
|
||||
version: schema.version,
|
||||
@@ -369,12 +381,79 @@ fn save_operation(
|
||||
tx.execute("INSERT INTO plugin_operations(workspace_id,operation_id,actor,kind,operation_key,fingerprint,result_json) VALUES(?1,?2,?3,?4,?5,?6,?7)",params![workspace,operation,actor,kind_name(kind),key,fingerprint,encode(&result)?])?;
|
||||
Ok(result)
|
||||
}
|
||||
// Collection/schema-id/version is a durable namespace within one installation.
|
||||
// Revocation does not erase that contract: a later artifact must use a new schema
|
||||
// version instead of reinterpreting retained cells under a changed schema.
|
||||
fn canonical_schema(value: &serde_json::Value) -> Result<String> {
|
||||
fn sorted(value: &serde_json::Value) -> serde_json::Value {
|
||||
match value {
|
||||
serde_json::Value::Object(map) => {
|
||||
let ordered: std::collections::BTreeMap<_, _> = map
|
||||
.iter()
|
||||
.map(|(key, value)| (key.clone(), sorted(value)))
|
||||
.collect();
|
||||
serde_json::Value::Object(ordered.into_iter().collect())
|
||||
}
|
||||
serde_json::Value::Array(values) => {
|
||||
serde_json::Value::Array(values.iter().map(sorted).collect())
|
||||
}
|
||||
value => value.clone(),
|
||||
}
|
||||
}
|
||||
encode(&sorted(value))
|
||||
}
|
||||
fn compare_data_schemas(
|
||||
tx: &Transaction<'_>,
|
||||
workspace: &str,
|
||||
installation: &str,
|
||||
artifact: &PluginArtifact,
|
||||
) -> Result<()> {
|
||||
let mut incoming = std::collections::BTreeMap::new();
|
||||
for schema in &artifact.data_collections {
|
||||
let address = (
|
||||
schema.collection.clone(),
|
||||
schema.schema_id.clone(),
|
||||
schema.schema_version.clone(),
|
||||
);
|
||||
let canonical = canonical_schema(&schema.schema)?;
|
||||
if let Some(previous) = incoming.insert(address, canonical.clone()) {
|
||||
if previous != canonical {
|
||||
return fail(PluginCatalogError::Conflict);
|
||||
}
|
||||
}
|
||||
}
|
||||
if incoming.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
// No revoked filter: the schema identity outlives execution authority.
|
||||
let mut statement = tx.prepare(
|
||||
"SELECT metadata_json FROM plugin_artifacts WHERE workspace_id=?1 AND installation_id=?2",
|
||||
)?;
|
||||
let mut rows = statement.query(params![workspace, installation])?;
|
||||
while let Some(row) = rows.next()? {
|
||||
let registered: PluginArtifact = decode(&row.get::<_, String>(0)?)?;
|
||||
for schema in registered.data_collections {
|
||||
let address = (
|
||||
schema.collection.clone(),
|
||||
schema.schema_id.clone(),
|
||||
schema.schema_version.clone(),
|
||||
);
|
||||
if let Some(candidate) = incoming.get(&address) {
|
||||
if *candidate != canonical_schema(&schema.schema)? {
|
||||
return fail(PluginCatalogError::Conflict);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn add_artifact(
|
||||
tx: &Transaction<'_>,
|
||||
workspace: &str,
|
||||
installation: &str,
|
||||
artifact: &PluginArtifact,
|
||||
) -> Result<()> {
|
||||
compare_data_schemas(tx, workspace, installation, artifact)?;
|
||||
let existing=tx.query_row("SELECT artifact_digest,package_digest,revoked FROM plugin_artifacts WHERE workspace_id=?1 AND installation_id=?2 AND version=?3",params![workspace,installation,artifact.version],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,bool>(2)?))).optional()?;
|
||||
if let Some((old, package, revoked)) = existing {
|
||||
if old != artifact.artifact_digest || package != artifact.package_digest || revoked {
|
||||
@@ -660,6 +739,8 @@ pub enum PluginReferenceKind {
|
||||
Pending,
|
||||
Active,
|
||||
Restorable,
|
||||
/// Retained typed extension data, including delete tombstones.
|
||||
Data,
|
||||
}
|
||||
impl PluginReferenceKind {
|
||||
fn name(self) -> &'static str {
|
||||
@@ -668,6 +749,7 @@ impl PluginReferenceKind {
|
||||
Self::Pending => "pending",
|
||||
Self::Active => "active",
|
||||
Self::Restorable => "restorable",
|
||||
Self::Data => "data",
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -729,5 +811,9 @@ pub fn plugin_reference_remove(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "plugin_data/catalog_tests.rs"]
|
||||
mod data_schema_tests;
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
@@ -30,7 +30,7 @@ fn plugin_catalog_migration_matches_fresh_schema_from_oldest_retained_and_105()
|
||||
// Plugin catalog remains migration 106 even when startup applies later steps.
|
||||
assert_eq!(after.iter().filter(|(version, _)| *version == 106).cloned().collect::<Vec<_>>(),
|
||||
vec![(106, crate::plugin_catalog::MIGRATION_NAME.into())]);
|
||||
assert_eq!(after.last().unwrap(), &(110, "Worker Plugin launch selection associations".into()));
|
||||
assert_eq!(after.last().unwrap(), &(111, "Typed Plugin Ticket extension data and atomic mutation receipts".into()));
|
||||
assert_eq!(current_schema_version(c)?, LATEST_SCHEMA_VERSION);
|
||||
assert_eq!(c.query_row("SELECT display_name FROM workspaces WHERE workspace_id='plugin-workspace'",[],|r|r.get::<_,String>(0))?,"Plugins");
|
||||
assert_eq!(c.query_row("SELECT count(*) FROM pragma_foreign_key_check",[],|r|r.get::<_,i64>(0))?,0);
|
||||
@@ -77,8 +77,12 @@ fn plugin_catalog_migration_marker_failure_rolls_back_then_retries_without_chang
|
||||
"Durable Plugin Host invocations and HTTPS effect receipts".into(),
|
||||
));
|
||||
expected.push((110, "Worker Plugin launch selection associations".into()));
|
||||
expected.push((
|
||||
111,
|
||||
"Typed Plugin Ticket extension data and atomic mutation receipts".into(),
|
||||
));
|
||||
assert_eq!(workspace_schema_migration_history(c)?, expected);
|
||||
crate::plugin_catalog::verify(c)?;
|
||||
crate::plugin_catalog::verify_with_data(c)?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
@@ -0,0 +1,600 @@
|
||||
//! Transactional Ticket extension cells, not an authorization or schema authority.
|
||||
//!
|
||||
//! The owner must reauthorize the exact object/collection on every call (including
|
||||
//! receipt replay), validate values against the pinned closed schema, and supply
|
||||
//! canonical Ticket and catalog identities. Use the Ticket owner's SQLite DB and
|
||||
//! an IMMEDIATE transaction; commit before reporting success. No guest identity,
|
||||
//! SQL, migration or cross-object transaction is exposed by this module.
|
||||
//!
|
||||
//! Cells and delete tombstones survive Ticket deletion and revocation. A parent
|
||||
//! existence check under write ownership plus SQL triggers fences deletion; the
|
||||
//! Workspace `deleting` state is also fenced. Explicit admin purge first requires
|
||||
//! all artifacts revoked, then compares the ordered exact cell/last-write set.
|
||||
use crate::{
|
||||
Error, Result,
|
||||
plugin_catalog::{self, PluginReference, PluginReferenceKind},
|
||||
plugin_receipt::{self, InvocationRecord},
|
||||
};
|
||||
use rusqlite::{Connection, OptionalExtension, Transaction, params};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use server_api::{
|
||||
PluginDataAddress, PluginDataCell, PluginDataCollection, PluginDataEntry,
|
||||
PluginDataExpectation, PluginDataMutation, PluginDataMutationLookup, PluginDataPage,
|
||||
PluginDataReceipt, PluginEffectObservation, PluginHostError, PluginHostErrorCode,
|
||||
};
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
pub const SCHEMA_SQL: &str = include_str!("plugin_data/schema.sql");
|
||||
/// Apply exactly once in a new ordered migration, before `SCHEMA_SQL`.
|
||||
pub const REFERENCE_MIGRATION_SQL: &str = include_str!("plugin_data/reference_migration.sql");
|
||||
const CURSOR_LIFETIME_MS: i64 = 10 * 60 * 1000;
|
||||
const MAX_CURSORS_PER_WORKSPACE: i64 = 4096;
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct Namespace {
|
||||
pub workspace_id: String,
|
||||
pub installation_id: String,
|
||||
pub ticket_id: String,
|
||||
pub collection: PluginDataCollection,
|
||||
}
|
||||
fn fail<T>(code: PluginHostErrorCode) -> Result<T> {
|
||||
// Key/identity conflicts and expired/cancelled invocations can have an older
|
||||
// committed mutation. Match the shared receipt owner's conservative signal.
|
||||
let effect = match code {
|
||||
PluginHostErrorCode::Conflict
|
||||
| PluginHostErrorCode::ReceiptExpired
|
||||
| PluginHostErrorCode::Cancelled => PluginEffectObservation::Indeterminate,
|
||||
_ => PluginEffectObservation::NotDispatched,
|
||||
};
|
||||
Err(PluginHostError {
|
||||
code,
|
||||
effect,
|
||||
diagnostic_id: uuid::Uuid::now_v7().to_string(),
|
||||
}
|
||||
.into())
|
||||
}
|
||||
fn valid(value: &str, max: usize) -> bool {
|
||||
!value.is_empty() && value.len() <= max && !value.chars().any(char::is_control)
|
||||
}
|
||||
fn encode(value: &impl Serialize) -> Result<String> {
|
||||
serde_json::to_string(value).map_err(|_| Error::Store("invalid Plugin data encoding".into()))
|
||||
}
|
||||
fn decode<T: serde::de::DeserializeOwned>(value: &str) -> Result<T> {
|
||||
serde_json::from_str(value).map_err(|_| Error::Store("invalid persisted Plugin data".into()))
|
||||
}
|
||||
fn hex(bytes: &[u8]) -> String {
|
||||
bytes.iter().map(|byte| format!("{byte:02x}")).collect()
|
||||
}
|
||||
fn hash(bytes: &[u8]) -> String {
|
||||
format!("sha256:{}", hex(&Sha256::digest(bytes)))
|
||||
}
|
||||
// Recursion is bounded by serde_json's parser; objects are explicitly sorted even
|
||||
// if another dependency enables serde_json/preserve_order through feature union.
|
||||
fn canonical(value: serde_json::Value) -> serde_json::Value {
|
||||
match value {
|
||||
serde_json::Value::Object(map) => {
|
||||
let sorted: std::collections::BTreeMap<_, _> = map
|
||||
.into_iter()
|
||||
.map(|(key, value)| (key, canonical(value)))
|
||||
.collect();
|
||||
serde_json::Value::Object(sorted.into_iter().collect())
|
||||
}
|
||||
serde_json::Value::Array(values) => {
|
||||
serde_json::Value::Array(values.into_iter().map(canonical).collect())
|
||||
}
|
||||
other => other,
|
||||
}
|
||||
}
|
||||
fn canonical_json(value: &str) -> Result<String> {
|
||||
if value.len() > server_api::PLUGIN_DATA_VALUE_LIMIT {
|
||||
return fail(PluginHostErrorCode::QuotaExceeded);
|
||||
}
|
||||
let value = serde_json::from_str(value).map_err(|_| {
|
||||
Error::from(PluginHostError {
|
||||
code: PluginHostErrorCode::InvalidInput,
|
||||
effect: PluginEffectObservation::NotDispatched,
|
||||
diagnostic_id: uuid::Uuid::now_v7().to_string(),
|
||||
})
|
||||
})?;
|
||||
let encoded = encode(&canonical(value))?;
|
||||
if encoded.len() > server_api::PLUGIN_DATA_VALUE_LIMIT {
|
||||
return fail(PluginHostErrorCode::QuotaExceeded);
|
||||
}
|
||||
Ok(encoded)
|
||||
}
|
||||
fn digest(value: &impl Serialize) -> Result<String> {
|
||||
let value =
|
||||
serde_json::to_value(value).map_err(|_| Error::Store("invalid Plugin data".into()))?;
|
||||
Ok(hash(encode(&canonical(value))?.as_bytes()))
|
||||
}
|
||||
fn validate_namespace(ns: &Namespace) -> Result<()> {
|
||||
if !valid(&ns.workspace_id, 512)
|
||||
|| !valid(&ns.installation_id, 512)
|
||||
|| !valid(&ns.ticket_id, 512)
|
||||
|| !valid(&ns.collection.collection, 128)
|
||||
|| !valid(&ns.collection.schema_id, 128)
|
||||
|| !valid(&ns.collection.schema_version, 128)
|
||||
{
|
||||
return fail(PluginHostErrorCode::InvalidInput);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn parent(conn: &Connection, ns: &Namespace) -> Result<()> {
|
||||
validate_namespace(ns)?;
|
||||
if !conn.query_row("SELECT EXISTS(SELECT 1 FROM typed_tickets t JOIN workspaces w USING(workspace_id) JOIN plugin_installations i ON i.workspace_id=t.workspace_id WHERE t.workspace_id=?1 AND t.ticket_id=?2 AND w.state='active' AND i.installation_id=?3 AND i.uninstalled=0)",params![ns.workspace_id,ns.ticket_id,ns.installation_id], |row| row.get::<_,bool>(0))? {
|
||||
return fail(PluginHostErrorCode::NotFound);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn invocation(
|
||||
conn: &Connection,
|
||||
record: &InvocationRecord,
|
||||
ns: &Namespace,
|
||||
now_ms: i64,
|
||||
) -> Result<InvocationRecord> {
|
||||
parent(conn, ns)?;
|
||||
let record = plugin_receipt::verify_record(conn, record)?;
|
||||
if record.workspace_id != ns.workspace_id || record.installation_id != ns.installation_id {
|
||||
return fail(PluginHostErrorCode::Conflict);
|
||||
}
|
||||
if now_ms < 0 {
|
||||
return fail(PluginHostErrorCode::InvalidInput);
|
||||
}
|
||||
Ok(record)
|
||||
}
|
||||
fn expired(record: &InvocationRecord, now_ms: i64) -> bool {
|
||||
record.expired || now_ms >= record.expires_at_ms
|
||||
}
|
||||
fn active(record: &InvocationRecord, now_ms: i64) -> Result<()> {
|
||||
if expired(record, now_ms) {
|
||||
return fail(PluginHostErrorCode::ReceiptExpired);
|
||||
}
|
||||
if record.cancelled {
|
||||
return fail(PluginHostErrorCode::Cancelled);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn validate_key(key: &str) -> Result<()> {
|
||||
if !valid(key, 256) {
|
||||
return fail(PluginHostErrorCode::InvalidInput);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn cell(conn: &Connection, ns: &Namespace, key: &str) -> Result<PluginDataCell> {
|
||||
let saved = conn.query_row("SELECT last_write,value_json FROM plugin_data_cells WHERE workspace_id=?1 AND installation_id=?2 AND ticket_id=?3 AND collection=?4 AND schema_id=?5 AND schema_version=?6 AND cell_key=?7",
|
||||
params![ns.workspace_id,ns.installation_id,ns.ticket_id,ns.collection.collection,ns.collection.schema_id,ns.collection.schema_version,key],
|
||||
|row| Ok((row.get::<_,String>(0)?,row.get::<_,Option<String>>(1)?))).optional()?;
|
||||
Ok(match saved {
|
||||
None => PluginDataCell::NeverWritten,
|
||||
Some((last_write, Some(value_json))) => PluginDataCell::Present {
|
||||
value_json,
|
||||
last_write,
|
||||
},
|
||||
Some((last_delete, None)) => PluginDataCell::Deleted { last_delete },
|
||||
})
|
||||
}
|
||||
/// Current read, never a receipt replay. Parent must check the exact read grant.
|
||||
pub fn get(conn: &Connection, ns: &Namespace, key: &str) -> Result<PluginDataCell> {
|
||||
parent(conn, ns)?;
|
||||
validate_key(key)?;
|
||||
cell(conn, ns, key)
|
||||
}
|
||||
|
||||
// Persist only a bounded selector, never authorization or a snapshot of cells.
|
||||
// The 256-bit CSPRNG token cannot be forged by editing namespace/offset/deadline.
|
||||
fn issue_cursor(
|
||||
conn: &Connection,
|
||||
ns: &Namespace,
|
||||
scope: &str,
|
||||
after: &str,
|
||||
deadline: i64,
|
||||
) -> Result<String> {
|
||||
use ring::rand::{SecureRandom, SystemRandom};
|
||||
if let Some(token) = conn.query_row("SELECT cursor_token FROM plugin_data_cursors WHERE workspace_id=?1 AND namespace_digest=?2 AND after_key=?3 AND expires_at_ms=?4",params![ns.workspace_id,scope,after,deadline],|row|row.get(0)).optional()? {
|
||||
return Ok(token);
|
||||
}
|
||||
let mut bytes = [0u8; 32];
|
||||
SystemRandom::new()
|
||||
.fill(&mut bytes)
|
||||
.map_err(|_| Error::Store("Plugin data cursor entropy unavailable".into()))?;
|
||||
let token = format!("pdc-{}", hex(&bytes));
|
||||
// The bounded INSERT predicate is evaluated under SQLite write ownership,
|
||||
// even when tests call list on a Connection rather than the owner's tx.
|
||||
let changed = conn.execute("INSERT INTO plugin_data_cursors(workspace_id,installation_id,cursor_token,namespace_digest,after_key,expires_at_ms) SELECT ?1,?2,?3,?4,?5,?6 WHERE (SELECT count(*) FROM plugin_data_cursors WHERE workspace_id=?1)<?7 ON CONFLICT(workspace_id,namespace_digest,after_key,expires_at_ms) DO NOTHING",params![ns.workspace_id,ns.installation_id,token,scope,after,deadline,MAX_CURSORS_PER_WORKSPACE])?;
|
||||
if changed == 1 {
|
||||
return Ok(token);
|
||||
}
|
||||
match conn.query_row("SELECT cursor_token FROM plugin_data_cursors WHERE workspace_id=?1 AND namespace_digest=?2 AND after_key=?3 AND expires_at_ms=?4",params![ns.workspace_id,scope,after,deadline],|row|row.get(0)).optional()? {
|
||||
Some(token)=>Ok(token),None=>fail(PluginHostErrorCode::QuotaExceeded),
|
||||
}
|
||||
}
|
||||
/// Stable bytewise key order including tombstones, not a cross-page snapshot.
|
||||
/// Opaque persisted cursors bind namespace/offset/expiry, not access authority.
|
||||
/// Current authorization is required; the owner commits cursor issuance in its
|
||||
/// IMMEDIATE transaction before returning a page. Expired selectors are pruned.
|
||||
pub fn list(
|
||||
conn: &Connection,
|
||||
ns: &Namespace,
|
||||
cursor: Option<&str>,
|
||||
limit: u32,
|
||||
now_ms: i64,
|
||||
) -> Result<PluginDataPage> {
|
||||
parent(conn, ns)?;
|
||||
if !(1..=server_api::PLUGIN_DATA_LIST_LIMIT).contains(&limit) || now_ms < 0 {
|
||||
return fail(PluginHostErrorCode::InvalidInput);
|
||||
}
|
||||
let scope = digest(ns)?;
|
||||
conn.execute(
|
||||
"DELETE FROM plugin_data_cursors WHERE workspace_id=?1 AND expires_at_ms<=?2",
|
||||
params![ns.workspace_id, now_ms],
|
||||
)?;
|
||||
let (after, deadline) = match cursor {
|
||||
Some(raw) => {
|
||||
if raw.len() != 68
|
||||
|| !raw.starts_with("pdc-")
|
||||
|| !raw[4..]
|
||||
.bytes()
|
||||
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
|
||||
{
|
||||
return fail(PluginHostErrorCode::InvalidInput);
|
||||
}
|
||||
let saved = conn.query_row("SELECT after_key,expires_at_ms FROM plugin_data_cursors WHERE workspace_id=?1 AND installation_id=?2 AND namespace_digest=?3 AND cursor_token=?4 AND expires_at_ms>?5",params![ns.workspace_id,ns.installation_id,scope,raw,now_ms],|row|Ok((row.get::<_,String>(0)?,row.get::<_,i64>(1)?))).optional()?;
|
||||
match saved {
|
||||
Some(value) => value,
|
||||
None => return fail(PluginHostErrorCode::InvalidInput),
|
||||
}
|
||||
}
|
||||
None => (
|
||||
String::new(),
|
||||
now_ms
|
||||
.checked_add(CURSOR_LIFETIME_MS)
|
||||
.ok_or_else(|| Error::InvalidInput("Plugin cursor time".into()))?,
|
||||
),
|
||||
};
|
||||
let mut stmt = conn.prepare("SELECT cell_key,last_write,value_json FROM plugin_data_cells WHERE workspace_id=?1 AND installation_id=?2 AND ticket_id=?3 AND collection=?4 AND schema_id=?5 AND schema_version=?6 AND cell_key>?7 COLLATE BINARY ORDER BY cell_key COLLATE BINARY LIMIT ?8")?;
|
||||
let mut entries = stmt
|
||||
.query_map(
|
||||
params![
|
||||
ns.workspace_id,
|
||||
ns.installation_id,
|
||||
ns.ticket_id,
|
||||
ns.collection.collection,
|
||||
ns.collection.schema_id,
|
||||
ns.collection.schema_version,
|
||||
after,
|
||||
i64::from(limit) + 1
|
||||
],
|
||||
|row| {
|
||||
let key = row.get(0)?;
|
||||
let last = row.get::<_, String>(1)?;
|
||||
let value = row.get::<_, Option<String>>(2)?;
|
||||
Ok(PluginDataEntry {
|
||||
key,
|
||||
cell: match value {
|
||||
Some(value_json) => PluginDataCell::Present {
|
||||
value_json,
|
||||
last_write: last,
|
||||
},
|
||||
None => PluginDataCell::Deleted { last_delete: last },
|
||||
},
|
||||
})
|
||||
},
|
||||
)?
|
||||
.collect::<rusqlite::Result<Vec<_>>>()?;
|
||||
let next_cursor = if entries.len() > limit as usize {
|
||||
entries.pop();
|
||||
Some(issue_cursor(
|
||||
conn,
|
||||
ns,
|
||||
&scope,
|
||||
&entries.last().expect("nonzero limit").key,
|
||||
deadline,
|
||||
)?)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
Ok(PluginDataPage {
|
||||
entries,
|
||||
next_cursor,
|
||||
})
|
||||
}
|
||||
|
||||
struct Saved {
|
||||
namespace: String,
|
||||
fingerprint: String,
|
||||
receipt: PluginDataReceipt,
|
||||
}
|
||||
fn saved(
|
||||
conn: &Connection,
|
||||
record: &InvocationRecord,
|
||||
operation_key: &str,
|
||||
) -> Result<Option<Saved>> {
|
||||
Ok(conn.query_row("SELECT namespace_json,fingerprint,mutation_id,result_digest FROM plugin_data_receipts WHERE workspace_id=?1 AND invocation_id=?2 AND operation_key=?3",
|
||||
params![record.workspace_id,record.invocation_id,operation_key], |row| Ok(Saved{namespace:row.get(0)?,fingerprint:row.get(1)?,receipt:PluginDataReceipt{mutation_id:row.get(2)?,result_digest:row.get(3)?}})).optional()?)
|
||||
}
|
||||
fn http_key(conn: &Connection, record: &InvocationRecord, key: &str) -> Result<bool> {
|
||||
Ok(conn.query_row("SELECT EXISTS(SELECT 1 FROM plugin_effect_receipts WHERE workspace_id=?1 AND invocation_id=?2 AND effect_key=?3)",params![record.workspace_id,record.invocation_id,key],|r|r.get(0))?)
|
||||
}
|
||||
fn address_json(ns: &Namespace, key: &str) -> Result<String> {
|
||||
encode(&(ns, key))
|
||||
}
|
||||
/// Lookup is scoped to one exact address; transport/DB errors are not NotRecorded.
|
||||
/// Expiry takes precedence over unused keys, as for the existing HTTPS receipts.
|
||||
pub fn lookup_mutation(
|
||||
conn: &Connection,
|
||||
record: &InvocationRecord,
|
||||
ns: &Namespace,
|
||||
key: &str,
|
||||
operation_key: &str,
|
||||
now_ms: i64,
|
||||
) -> Result<PluginDataMutationLookup> {
|
||||
validate_key(key)?;
|
||||
validate_key(operation_key)?;
|
||||
let record = invocation(conn, record, ns, now_ms)?;
|
||||
if expired(&record, now_ms) {
|
||||
return Ok(PluginDataMutationLookup::Expired);
|
||||
}
|
||||
if record.cancelled {
|
||||
return fail(PluginHostErrorCode::Cancelled);
|
||||
}
|
||||
if let Some(saved) = saved(conn, &record, operation_key)? {
|
||||
if saved.namespace != address_json(ns, key)? {
|
||||
return fail(PluginHostErrorCode::Conflict);
|
||||
}
|
||||
return Ok(PluginDataMutationLookup::Committed(saved.receipt));
|
||||
}
|
||||
if http_key(conn, &record, operation_key)? {
|
||||
return fail(PluginHostErrorCode::Conflict);
|
||||
}
|
||||
Ok(PluginDataMutationLookup::NotRecorded)
|
||||
}
|
||||
fn atomic<T>(tx: &Transaction<'_>, work: impl FnOnce() -> Result<T>) -> Result<T> {
|
||||
tx.execute_batch("SAVEPOINT plugin_data_atomic")?;
|
||||
match work() {
|
||||
Ok(value) => {
|
||||
tx.execute_batch("RELEASE plugin_data_atomic")?;
|
||||
Ok(value)
|
||||
}
|
||||
Err(error) => {
|
||||
tx.execute_batch("ROLLBACK TO plugin_data_atomic; RELEASE plugin_data_atomic")?;
|
||||
Err(error)
|
||||
}
|
||||
}
|
||||
}
|
||||
/// Put/delete and receipt are atomic even if the caller catches an error and
|
||||
/// commits unrelated work. `reference` is Host-admitted, never guest input.
|
||||
/// Replay is checked before catalog/CAS and returns the original receipt only.
|
||||
pub fn mutate(
|
||||
tx: &Transaction<'_>,
|
||||
record: &InvocationRecord,
|
||||
ns: &Namespace,
|
||||
key: &str,
|
||||
operation_key: &str,
|
||||
mutation: &PluginDataMutation,
|
||||
reference: &PluginReference,
|
||||
now_ms: i64,
|
||||
) -> Result<PluginDataReceipt> {
|
||||
validate_key(key)?;
|
||||
validate_key(operation_key)?;
|
||||
let record = invocation(tx, record, ns, now_ms)?;
|
||||
active(&record, now_ms)?;
|
||||
let mutation = match mutation {
|
||||
PluginDataMutation::Put {
|
||||
expected,
|
||||
value_json,
|
||||
} => PluginDataMutation::Put {
|
||||
expected: expected.clone(),
|
||||
value_json: canonical_json(value_json)?,
|
||||
},
|
||||
PluginDataMutation::Delete {
|
||||
expected_last_write,
|
||||
} => PluginDataMutation::Delete {
|
||||
expected_last_write: expected_last_write.clone(),
|
||||
},
|
||||
};
|
||||
let address = address_json(ns, key)?;
|
||||
let fingerprint = digest(&("ticket-data-mutation", ns, key, &mutation))?;
|
||||
if let Some(saved) = saved(tx, &record, operation_key)? {
|
||||
if saved.namespace != address || saved.fingerprint != fingerprint {
|
||||
return fail(PluginHostErrorCode::Conflict);
|
||||
}
|
||||
return Ok(saved.receipt);
|
||||
}
|
||||
if http_key(tx, &record, operation_key)? {
|
||||
return fail(PluginHostErrorCode::Conflict);
|
||||
}
|
||||
let expected = match &mutation {
|
||||
PluginDataMutation::Put { expected, .. } => expected.clone(),
|
||||
PluginDataMutation::Delete {
|
||||
expected_last_write,
|
||||
} => PluginDataExpectation::LastWrite(expected_last_write.clone()),
|
||||
};
|
||||
if let PluginDataExpectation::LastWrite(last) = &expected {
|
||||
if !valid(last, 512) {
|
||||
return fail(PluginHostErrorCode::InvalidInput);
|
||||
}
|
||||
}
|
||||
atomic(tx, || {
|
||||
let observed = cell(tx, ns, key)?;
|
||||
let matches = match (&expected, &observed) {
|
||||
(PluginDataExpectation::NeverWritten, PluginDataCell::NeverWritten) => true,
|
||||
(
|
||||
PluginDataExpectation::LastWrite(want),
|
||||
PluginDataCell::Present { last_write, .. },
|
||||
) => want == last_write,
|
||||
(PluginDataExpectation::LastWrite(want), PluginDataCell::Deleted { last_delete }) => {
|
||||
want == last_delete
|
||||
}
|
||||
_ => false,
|
||||
};
|
||||
if !matches {
|
||||
return fail(PluginHostErrorCode::Conflict);
|
||||
}
|
||||
if reference.installation_id != ns.installation_id {
|
||||
return fail(PluginHostErrorCode::Conflict);
|
||||
}
|
||||
let owner = digest(ns)?;
|
||||
let reference = PluginReference {
|
||||
kind: PluginReferenceKind::Data,
|
||||
owner: owner.clone(),
|
||||
..reference.clone()
|
||||
};
|
||||
plugin_catalog::plugin_reference_put(tx, &ns.workspace_id, &reference)?;
|
||||
tx.execute("INSERT INTO plugin_data_holders(workspace_id,installation_id,owner,artifact_digest) VALUES(?1,?2,?3,?4) ON CONFLICT DO NOTHING",params![ns.workspace_id,ns.installation_id,owner,reference.artifact_digest])?;
|
||||
let mutation_id = uuid::Uuid::now_v7().to_string();
|
||||
let value = match &mutation {
|
||||
PluginDataMutation::Put { value_json, .. } => Some(value_json.as_str()),
|
||||
PluginDataMutation::Delete { .. } => None,
|
||||
};
|
||||
tx.execute("INSERT INTO plugin_data_cells(workspace_id,installation_id,ticket_id,collection,schema_id,schema_version,cell_key,last_write,value_json) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9) ON CONFLICT(workspace_id,installation_id,ticket_id,collection,schema_id,schema_version,cell_key) DO UPDATE SET last_write=excluded.last_write,value_json=excluded.value_json",
|
||||
params![ns.workspace_id,ns.installation_id,ns.ticket_id,ns.collection.collection,ns.collection.schema_id,ns.collection.schema_version,key,mutation_id,value])?;
|
||||
let result = match value {
|
||||
Some(value_json) => PluginDataCell::Present {
|
||||
value_json: value_json.into(),
|
||||
last_write: mutation_id.clone(),
|
||||
},
|
||||
None => PluginDataCell::Deleted {
|
||||
last_delete: mutation_id.clone(),
|
||||
},
|
||||
};
|
||||
let receipt = PluginDataReceipt {
|
||||
mutation_id,
|
||||
result_digest: digest(&result)?,
|
||||
};
|
||||
tx.execute("INSERT INTO plugin_data_receipts(workspace_id,invocation_id,operation_key,namespace_json,fingerprint,mutation_id,result_digest,result_json) VALUES(?1,?2,?3,?4,?5,?6,?7,?8)",
|
||||
params![ns.workspace_id,record.invocation_id,operation_key,address,fingerprint,receipt.mutation_id,receipt.result_digest,encode(&result)?])?;
|
||||
Ok(receipt)
|
||||
})
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct PurgeSnapshot {
|
||||
pub installation_last_operation: String,
|
||||
pub cells_digest: String,
|
||||
pub cell_count: u64,
|
||||
}
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct PurgeReceipt {
|
||||
pub purge_id: String,
|
||||
pub observed: PurgeSnapshot,
|
||||
}
|
||||
fn purge_fenced(conn: &Connection, workspace: &str, installation: &str) -> Result<String> {
|
||||
if !valid(workspace, 512) || !valid(installation, 512) {
|
||||
return fail(PluginHostErrorCode::InvalidInput);
|
||||
}
|
||||
let operation = conn.query_row("SELECT i.last_operation FROM plugin_installations i JOIN workspaces w USING(workspace_id) WHERE i.workspace_id=?1 AND i.installation_id=?2 AND i.uninstalled=0 AND w.state='active' AND NOT EXISTS(SELECT 1 FROM plugin_artifacts a WHERE a.workspace_id=i.workspace_id AND a.installation_id=i.installation_id AND a.revoked=0)",params![workspace,installation],|row|row.get(0)).optional()?;
|
||||
match operation {
|
||||
Some(value) => Ok(value),
|
||||
None => fail(PluginHostErrorCode::Conflict),
|
||||
}
|
||||
}
|
||||
/// Observe only after execution is fenced by catalog revocation. Hash a streamed,
|
||||
/// ordered set of exact namespace/key/last-write/deleted tuples, not a generation.
|
||||
/// Caller keeps the same IMMEDIATE transaction through this observation.
|
||||
pub fn purge_snapshot(
|
||||
tx: &Transaction<'_>,
|
||||
workspace: &str,
|
||||
installation: &str,
|
||||
) -> Result<PurgeSnapshot> {
|
||||
let installation_last_operation = purge_fenced(tx, workspace, installation)?;
|
||||
let mut stmt = tx.prepare("SELECT ticket_id,collection,schema_id,schema_version,cell_key,last_write,value_json IS NULL FROM plugin_data_cells WHERE workspace_id=?1 AND installation_id=?2 ORDER BY ticket_id COLLATE BINARY,collection COLLATE BINARY,schema_id COLLATE BINARY,schema_version COLLATE BINARY,cell_key COLLATE BINARY")?;
|
||||
let mut rows = stmt.query(params![workspace, installation])?;
|
||||
let mut hash = Sha256::new();
|
||||
hash.update(b"ticket-data-purge-exact-set-v1\0");
|
||||
hash.update(encode(&(workspace, installation))?.as_bytes());
|
||||
let mut count = 0u64;
|
||||
while let Some(row) = rows.next()? {
|
||||
let tuple = (
|
||||
row.get::<_, String>(0)?,
|
||||
row.get::<_, String>(1)?,
|
||||
row.get::<_, String>(2)?,
|
||||
row.get::<_, String>(3)?,
|
||||
row.get::<_, String>(4)?,
|
||||
row.get::<_, String>(5)?,
|
||||
row.get::<_, bool>(6)?,
|
||||
);
|
||||
let bytes = encode(&tuple)?;
|
||||
hash.update((bytes.len() as u64).to_be_bytes());
|
||||
hash.update(bytes.as_bytes());
|
||||
count += 1;
|
||||
}
|
||||
Ok(PurgeSnapshot {
|
||||
installation_last_operation,
|
||||
cells_digest: format!("sha256:{}", hex(&hash.finalize())),
|
||||
cell_count: count,
|
||||
})
|
||||
}
|
||||
/// Current Workspace-admin auth is external and required even for replay. Purge
|
||||
/// receipts survive data removal and uninstall. No implicit uninstall or revoke.
|
||||
pub fn purge(
|
||||
tx: &Transaction<'_>,
|
||||
workspace: &str,
|
||||
installation: &str,
|
||||
actor: &str,
|
||||
operation_key: &str,
|
||||
expected: &PurgeSnapshot,
|
||||
) -> Result<PurgeReceipt> {
|
||||
if !valid(workspace, 512) || !valid(installation, 512) || !valid(actor, 512) {
|
||||
return fail(PluginHostErrorCode::InvalidInput);
|
||||
}
|
||||
validate_key(operation_key)?;
|
||||
let fingerprint = digest(&("ticket-data-purge", installation, expected))?;
|
||||
if let Some((saved_fingerprint,result)) = tx.query_row("SELECT fingerprint,result_json FROM plugin_data_purges WHERE workspace_id=?1 AND actor=?2 AND operation_key=?3",params![workspace,actor,operation_key],|row|Ok((row.get::<_,String>(0)?,row.get::<_,String>(1)?))).optional()? {
|
||||
if saved_fingerprint!=fingerprint { return fail(PluginHostErrorCode::Conflict); }
|
||||
return decode(&result);
|
||||
}
|
||||
atomic(tx, || {
|
||||
let observed = purge_snapshot(tx, workspace, installation)?;
|
||||
if observed != *expected {
|
||||
return fail(PluginHostErrorCode::Conflict);
|
||||
}
|
||||
let holders = tx.prepare("SELECT owner,artifact_digest FROM plugin_data_holders WHERE workspace_id=?1 AND installation_id=?2")?
|
||||
.query_map(params![workspace,installation],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?)))?.collect::<rusqlite::Result<Vec<_>>>()?;
|
||||
let receipt = PurgeReceipt {
|
||||
purge_id: uuid::Uuid::now_v7().to_string(),
|
||||
observed,
|
||||
};
|
||||
tx.execute("INSERT INTO plugin_data_purges(workspace_id,actor,operation_key,installation_id,fingerprint,result_json) VALUES(?1,?2,?3,?4,?5,?6)",params![workspace,actor,operation_key,installation,fingerprint,encode(&receipt)?])?;
|
||||
tx.execute(
|
||||
"DELETE FROM plugin_data_cells WHERE workspace_id=?1 AND installation_id=?2",
|
||||
params![workspace, installation],
|
||||
)?;
|
||||
tx.execute(
|
||||
"DELETE FROM plugin_data_cursors WHERE workspace_id=?1 AND installation_id=?2",
|
||||
params![workspace, installation],
|
||||
)?;
|
||||
for (owner, artifact) in holders {
|
||||
plugin_catalog::plugin_reference_remove(
|
||||
tx,
|
||||
workspace,
|
||||
PluginReferenceKind::Data,
|
||||
&owner,
|
||||
installation,
|
||||
&artifact,
|
||||
)?;
|
||||
}
|
||||
tx.execute(
|
||||
"DELETE FROM plugin_data_holders WHERE workspace_id=?1 AND installation_id=?2",
|
||||
params![workspace, installation],
|
||||
)?;
|
||||
Ok(receipt)
|
||||
})
|
||||
}
|
||||
|
||||
/// Convenience constructor for the owner after resolving a wire address.
|
||||
pub fn namespace(
|
||||
record: &InvocationRecord,
|
||||
ticket_id: &str,
|
||||
address: &PluginDataAddress,
|
||||
) -> Namespace {
|
||||
Namespace {
|
||||
workspace_id: record.workspace_id.clone(),
|
||||
installation_id: record.installation_id.clone(),
|
||||
ticket_id: ticket_id.into(),
|
||||
collection: address.collection.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
@@ -0,0 +1,159 @@
|
||||
//! Real SQLite tests of catalog artifact registration and immutable typed schema
|
||||
//! namespaces. Manifest/Component validation and caller authorization are outside
|
||||
//! this owner test; immutable metadata DTOs are the input boundary. No guest runs.
|
||||
use super::*;
|
||||
use std::{
|
||||
sync::{Arc, Barrier},
|
||||
time::Duration as StdDuration,
|
||||
};
|
||||
use tempfile::tempdir;
|
||||
|
||||
fn artifact(version: &str, schema_version: &str, schema: serde_json::Value) -> PluginArtifact {
|
||||
serde_json::from_value(serde_json::json!({
|
||||
"package_id":"package","name":"package","version":version,"description":null,
|
||||
"package_digest":format!("sha256:{}",hash(format!("package-{version}").as_bytes())),
|
||||
"artifact_digest":format!("sha256:{}",hash(format!("artifact-{version}").as_bytes())),"artifact_size":1,
|
||||
"schema_version":2,"world":manifest::plugin::PLUGIN_COMPONENT_TICKET_TOOL_WORLD,
|
||||
"required_imports":[],"tools":[],"connections":[],"requested_grants":[],
|
||||
"data_collections":[{"collection":"collection","schema_id":"schema","schema_version":schema_version,"schema":schema}],
|
||||
"config_schema":{"identity":"config","version":"1","namespace":"package","digest":format!("sha256:{}",hash(b"config")),"dcdl":""},
|
||||
"revoked":false,"revocation_reason":null
|
||||
})).unwrap()
|
||||
}
|
||||
fn schema(value_type: &str) -> serde_json::Value {
|
||||
serde_json::json!({"type":"object","properties":{"value":{"type":value_type}},"required":["value"],"additionalProperties":false})
|
||||
}
|
||||
fn open(path: &Path) -> Connection {
|
||||
let conn = Connection::open(path).unwrap();
|
||||
conn.execute_batch("PRAGMA foreign_keys=ON; PRAGMA journal_mode=WAL;")
|
||||
.unwrap();
|
||||
conn.busy_timeout(StdDuration::from_secs(10)).unwrap();
|
||||
conn
|
||||
}
|
||||
fn initialize(conn: &Connection) {
|
||||
conn.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY); INSERT INTO workspaces VALUES('w'),('other-workspace');").unwrap();
|
||||
conn.execute_batch(include_str!("../plugin_catalog_schema.sql"))
|
||||
.unwrap();
|
||||
for workspace in ["w", "other-workspace"] {
|
||||
for installation in ["i", "other-installation"] {
|
||||
conn.execute(
|
||||
"INSERT INTO plugin_installations VALUES(?1,?2,?2,'package','op',0)",
|
||||
params![workspace, installation],
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
fn register(
|
||||
conn: &mut Connection,
|
||||
workspace: &str,
|
||||
installation: &str,
|
||||
candidate: &PluginArtifact,
|
||||
) -> Result<()> {
|
||||
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
|
||||
add_artifact(&tx, workspace, installation, candidate)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
}
|
||||
fn conflict(result: Result<()>) {
|
||||
assert!(
|
||||
matches!(
|
||||
result,
|
||||
Err(Error::PluginCatalog(PluginCatalogError::Conflict))
|
||||
),
|
||||
"expected schema conflict, got {result:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn registered_schema_content_is_immutable_per_namespace_including_revoked_artifacts() {
|
||||
let dir = tempdir().unwrap();
|
||||
let path = dir.path().join("catalog.db");
|
||||
let mut conn = open(&path);
|
||||
initialize(&conn);
|
||||
let first = artifact("1.0.0", "1", schema("string"));
|
||||
register(&mut conn, "w", "i", &first).unwrap();
|
||||
// Different object member order is the same canonical schema, not a change.
|
||||
let reordered=serde_json::from_str(r#"{"additionalProperties":false,"required":["value"],"properties":{"value":{"type":"string"}},"type":"object"}"#).unwrap();
|
||||
register(&mut conn, "w", "i", &artifact("1.0.1", "1", reordered)).unwrap();
|
||||
conn.execute("UPDATE plugin_artifacts SET revoked=1,revocation_reason='admin' WHERE workspace_id='w' AND installation_id='i'",[]).unwrap();
|
||||
drop(conn);
|
||||
let mut conn = open(&path);
|
||||
let changed = artifact("2.0.0", "1", schema("integer"));
|
||||
conflict(register(&mut conn, "w", "i", &changed));
|
||||
assert_eq!(
|
||||
conn.query_row(
|
||||
"SELECT count(*) FROM plugin_artifacts WHERE workspace_id='w' AND installation_id='i'",
|
||||
[],
|
||||
|row| row.get::<_, i64>(0)
|
||||
)
|
||||
.unwrap(),
|
||||
2
|
||||
);
|
||||
// Explicit new schema version registers a separate namespace; rollback keeps
|
||||
// the original schema metadata instead of reading it as the new definition.
|
||||
let next = artifact("2.0.0", "2", schema("integer"));
|
||||
register(&mut conn, "w", "i", &next).unwrap();
|
||||
let saved = load_installation(&conn, "w", "i").unwrap();
|
||||
let old = saved
|
||||
.artifacts
|
||||
.iter()
|
||||
.find(|a| a.version == "1.0.0")
|
||||
.unwrap();
|
||||
let new = saved
|
||||
.artifacts
|
||||
.iter()
|
||||
.find(|a| a.version == "2.0.0")
|
||||
.unwrap();
|
||||
assert!(old.revoked);
|
||||
assert_eq!(old.data_collections[0].schema_version, "1");
|
||||
assert_eq!(old.data_collections[0].schema, schema("string"));
|
||||
assert_eq!(new.data_collections[0].schema_version, "2");
|
||||
assert_eq!(new.data_collections[0].schema, schema("integer"));
|
||||
// Installation and Workspace ownership are independent schema namespaces.
|
||||
register(&mut conn, "w", "other-installation", &changed).unwrap();
|
||||
register(&mut conn, "other-workspace", "i", &changed).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn concurrent_artifact_registration_cannot_redefine_a_schema_namespace() {
|
||||
let dir = tempdir().unwrap();
|
||||
let path = dir.path().join("catalog.db");
|
||||
initialize(&open(&path));
|
||||
let gate = Arc::new(Barrier::new(3));
|
||||
let handles: Vec<_> = [("1.0.0", "string"), ("2.0.0", "integer")]
|
||||
.into_iter()
|
||||
.map(|(version, kind)| {
|
||||
let path = path.clone();
|
||||
let gate = gate.clone();
|
||||
std::thread::spawn(move || {
|
||||
let mut conn = open(&path);
|
||||
gate.wait();
|
||||
let result = register(&mut conn, "w", "i", &artifact(version, "1", schema(kind)));
|
||||
match result {
|
||||
Ok(()) => true,
|
||||
Err(error) => {
|
||||
assert!(matches!(
|
||||
error,
|
||||
Error::PluginCatalog(PluginCatalogError::Conflict)
|
||||
));
|
||||
false
|
||||
}
|
||||
}
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
gate.wait();
|
||||
let winners = handles
|
||||
.into_iter()
|
||||
.map(|handle| usize::from(handle.join().unwrap()))
|
||||
.sum::<usize>();
|
||||
assert_eq!(winners, 1);
|
||||
assert_eq!(
|
||||
load_installation(&open(&path), "w", "i")
|
||||
.unwrap()
|
||||
.artifacts
|
||||
.len(),
|
||||
1
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
-- New ordered migration only. Never modify the published catalog baseline.
|
||||
DROP TRIGGER plugin_reference_live;
|
||||
DROP TRIGGER plugin_uninstall_referenced;
|
||||
CREATE TABLE plugin_references_with_data (
|
||||
workspace_id TEXT NOT NULL,
|
||||
kind TEXT NOT NULL CHECK(kind IN ('config','pending','active','restorable','data')),
|
||||
owner TEXT NOT NULL CHECK(length(owner) BETWEEN 1 AND 512),
|
||||
installation_id TEXT NOT NULL,
|
||||
artifact_digest TEXT NOT NULL,
|
||||
PRIMARY KEY(workspace_id, kind, owner, installation_id, artifact_digest),
|
||||
FOREIGN KEY(workspace_id, installation_id, artifact_digest) REFERENCES plugin_artifacts(workspace_id, installation_id, artifact_digest) ON DELETE RESTRICT
|
||||
);
|
||||
INSERT INTO plugin_references_with_data SELECT * FROM plugin_references;
|
||||
DROP TABLE plugin_references;
|
||||
ALTER TABLE plugin_references_with_data RENAME TO plugin_references;
|
||||
CREATE TRIGGER plugin_reference_live BEFORE INSERT ON plugin_references
|
||||
WHEN NOT EXISTS(SELECT 1 FROM plugin_artifacts a JOIN plugin_installations i USING(workspace_id, installation_id)
|
||||
WHERE a.workspace_id=NEW.workspace_id AND a.installation_id=NEW.installation_id AND a.artifact_digest=NEW.artifact_digest AND a.revoked=0 AND i.uninstalled=0)
|
||||
BEGIN SELECT RAISE(ABORT,'plugin reference target unavailable'); END;
|
||||
CREATE TRIGGER plugin_uninstall_referenced BEFORE UPDATE OF uninstalled ON plugin_installations
|
||||
WHEN NEW.uninstalled=1 AND EXISTS(SELECT 1 FROM plugin_references r WHERE r.workspace_id=NEW.workspace_id AND r.installation_id=NEW.installation_id)
|
||||
BEGIN SELECT RAISE(ABORT,'plugin installation referenced'); END;
|
||||
@@ -0,0 +1,105 @@
|
||||
-- Ticket children survive parent deletion for explicit Host retention/purge.
|
||||
-- No guest SQL access; caller and triggers share the Ticket owner's DB fence.
|
||||
CREATE TABLE plugin_data_cells (
|
||||
workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
|
||||
installation_id TEXT NOT NULL,
|
||||
ticket_id TEXT NOT NULL,
|
||||
collection TEXT NOT NULL,
|
||||
schema_id TEXT NOT NULL,
|
||||
schema_version TEXT NOT NULL,
|
||||
cell_key TEXT NOT NULL CHECK(length(CAST(cell_key AS BLOB)) BETWEEN 1 AND 512),
|
||||
last_write TEXT NOT NULL UNIQUE,
|
||||
value_json TEXT CHECK(value_json IS NULL OR (json_valid(value_json) AND length(CAST(value_json AS BLOB))<=65536)),
|
||||
PRIMARY KEY(workspace_id,installation_id,ticket_id,collection,schema_id,schema_version,cell_key),
|
||||
FOREIGN KEY(workspace_id,installation_id) REFERENCES plugin_installations(workspace_id,installation_id)
|
||||
);
|
||||
CREATE TRIGGER plugin_data_parent_insert BEFORE INSERT ON plugin_data_cells
|
||||
WHEN NOT EXISTS(SELECT 1 FROM typed_tickets t JOIN workspaces w USING(workspace_id)
|
||||
WHERE t.workspace_id=NEW.workspace_id AND t.ticket_id=NEW.ticket_id AND w.state='active')
|
||||
BEGIN SELECT RAISE(ABORT,'Plugin data parent unavailable'); END;
|
||||
CREATE TRIGGER plugin_data_parent_update BEFORE UPDATE ON plugin_data_cells
|
||||
WHEN NOT EXISTS(SELECT 1 FROM typed_tickets t JOIN workspaces w USING(workspace_id)
|
||||
WHERE t.workspace_id=NEW.workspace_id AND t.ticket_id=NEW.ticket_id AND w.state='active')
|
||||
BEGIN SELECT RAISE(ABORT,'Plugin data parent unavailable'); END;
|
||||
|
||||
-- Random opaque pagination selectors only, never authorization. Expired rows
|
||||
-- are discarded; API issuance is bounded per Workspace and cannot extend a
|
||||
-- traversal deadline. Explicit purge invalidates installation selectors.
|
||||
CREATE TABLE plugin_data_cursors (
|
||||
workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
|
||||
installation_id TEXT NOT NULL,
|
||||
cursor_token TEXT NOT NULL CHECK(length(cursor_token)=68 AND substr(cursor_token,1,4)='pdc-' AND substr(cursor_token,5) NOT GLOB '*[^0-9a-f]*'),
|
||||
namespace_digest TEXT NOT NULL CHECK(length(namespace_digest)=71),
|
||||
after_key TEXT NOT NULL CHECK(length(CAST(after_key AS BLOB)) BETWEEN 1 AND 256),
|
||||
expires_at_ms INTEGER NOT NULL CHECK(expires_at_ms>0),
|
||||
PRIMARY KEY(workspace_id,cursor_token),
|
||||
UNIQUE(workspace_id,namespace_digest,after_key,expires_at_ms),
|
||||
FOREIGN KEY(workspace_id,installation_id) REFERENCES plugin_installations(workspace_id,installation_id)
|
||||
);
|
||||
CREATE INDEX plugin_data_cursor_expiry ON plugin_data_cursors(workspace_id,expires_at_ms);
|
||||
|
||||
-- Conservatively retain each writing artifact until explicit installation purge,
|
||||
-- even when a later artifact overwrites its cell. owner is exact namespace hash.
|
||||
CREATE TABLE plugin_data_holders (
|
||||
workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
|
||||
installation_id TEXT NOT NULL,
|
||||
owner TEXT NOT NULL,
|
||||
artifact_digest TEXT NOT NULL,
|
||||
PRIMARY KEY(workspace_id,installation_id,owner,artifact_digest),
|
||||
FOREIGN KEY(workspace_id,installation_id,artifact_digest) REFERENCES plugin_artifacts(workspace_id,installation_id,artifact_digest)
|
||||
);
|
||||
CREATE TABLE plugin_data_receipts (
|
||||
workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
|
||||
invocation_id TEXT NOT NULL,
|
||||
operation_key TEXT NOT NULL CHECK(length(CAST(operation_key AS BLOB)) BETWEEN 1 AND 256),
|
||||
namespace_json TEXT NOT NULL CHECK(json_valid(namespace_json)),
|
||||
fingerprint TEXT NOT NULL CHECK(length(fingerprint)=71),
|
||||
mutation_id TEXT NOT NULL UNIQUE,
|
||||
result_digest TEXT NOT NULL CHECK(length(result_digest)=71),
|
||||
result_json TEXT CHECK(result_json IS NULL OR (json_valid(result_json) AND length(CAST(result_json AS BLOB))<=262144)),
|
||||
PRIMARY KEY(workspace_id,invocation_id,operation_key),
|
||||
FOREIGN KEY(workspace_id,invocation_id) REFERENCES plugin_invocations(workspace_id,invocation_id)
|
||||
);
|
||||
CREATE TRIGGER plugin_data_receipt_immutable BEFORE UPDATE ON plugin_data_receipts
|
||||
WHEN NOT(OLD.result_json IS NOT NULL AND NEW.result_json IS NULL
|
||||
AND NEW.workspace_id=OLD.workspace_id AND NEW.invocation_id=OLD.invocation_id
|
||||
AND NEW.operation_key=OLD.operation_key AND NEW.namespace_json=OLD.namespace_json
|
||||
AND NEW.fingerprint=OLD.fingerprint AND NEW.mutation_id=OLD.mutation_id AND NEW.result_digest=OLD.result_digest
|
||||
AND EXISTS(SELECT 1 FROM plugin_invocations i WHERE i.workspace_id=OLD.workspace_id AND i.invocation_id=OLD.invocation_id AND i.expired=1))
|
||||
BEGIN SELECT RAISE(ABORT,'Plugin data receipt immutable'); END;
|
||||
CREATE TRIGGER plugin_data_receipt_no_replace BEFORE INSERT ON plugin_data_receipts
|
||||
WHEN EXISTS(SELECT 1 FROM plugin_data_receipts WHERE (workspace_id=NEW.workspace_id AND invocation_id=NEW.invocation_id AND operation_key=NEW.operation_key) OR mutation_id=NEW.mutation_id)
|
||||
BEGIN SELECT RAISE(ABORT,'Plugin data operation key retained'); END;
|
||||
CREATE TRIGGER plugin_data_receipt_retained BEFORE DELETE ON plugin_data_receipts
|
||||
WHEN EXISTS(SELECT 1 FROM workspaces WHERE workspace_id=OLD.workspace_id)
|
||||
BEGIN SELECT RAISE(ABORT,'Plugin data receipt retained'); END;
|
||||
CREATE TRIGGER plugin_data_receipt_expiry AFTER UPDATE OF expired ON plugin_invocations
|
||||
WHEN NEW.expired=1 AND OLD.expired=0
|
||||
BEGIN UPDATE plugin_data_receipts SET result_json=NULL WHERE workspace_id=NEW.workspace_id AND invocation_id=NEW.invocation_id AND result_json IS NOT NULL; END;
|
||||
|
||||
CREATE TRIGGER plugin_effect_data_collision BEFORE INSERT ON plugin_effect_receipts
|
||||
WHEN EXISTS(SELECT 1 FROM plugin_data_receipts WHERE workspace_id=NEW.workspace_id AND invocation_id=NEW.invocation_id AND operation_key=NEW.effect_key)
|
||||
BEGIN SELECT RAISE(ABORT,'Plugin operation key belongs to data'); END;
|
||||
CREATE TRIGGER plugin_data_effect_collision BEFORE INSERT ON plugin_data_receipts
|
||||
WHEN EXISTS(SELECT 1 FROM plugin_effect_receipts WHERE workspace_id=NEW.workspace_id AND invocation_id=NEW.invocation_id AND effect_key=NEW.operation_key)
|
||||
BEGIN SELECT RAISE(ABORT,'Plugin operation key belongs to HTTP'); END;
|
||||
|
||||
-- Admin operations are not guest invocations; this is only an atomic purge
|
||||
-- receipt, never an execution identity or saved authorization decision.
|
||||
CREATE TABLE plugin_data_purges (
|
||||
workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
|
||||
actor TEXT NOT NULL,
|
||||
operation_key TEXT NOT NULL CHECK(length(CAST(operation_key AS BLOB)) BETWEEN 1 AND 256),
|
||||
installation_id TEXT NOT NULL,
|
||||
fingerprint TEXT NOT NULL,
|
||||
result_json TEXT NOT NULL CHECK(json_valid(result_json)),
|
||||
PRIMARY KEY(workspace_id,actor,operation_key)
|
||||
);
|
||||
CREATE TRIGGER plugin_data_purge_immutable BEFORE UPDATE ON plugin_data_purges
|
||||
BEGIN SELECT RAISE(ABORT,'Plugin purge receipt immutable'); END;
|
||||
CREATE TRIGGER plugin_data_purge_no_replace BEFORE INSERT ON plugin_data_purges
|
||||
WHEN EXISTS(SELECT 1 FROM plugin_data_purges WHERE workspace_id=NEW.workspace_id AND actor=NEW.actor AND operation_key=NEW.operation_key)
|
||||
BEGIN SELECT RAISE(ABORT,'Plugin purge key retained'); END;
|
||||
CREATE TRIGGER plugin_data_purge_retained BEFORE DELETE ON plugin_data_purges
|
||||
WHEN EXISTS(SELECT 1 FROM workspaces WHERE workspace_id=OLD.workspace_id)
|
||||
BEGIN SELECT RAISE(ABORT,'Plugin purge receipt retained'); END;
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,63 @@
|
||||
// The real retained predecessor exercises both the catalog CHECK rebuild and
|
||||
// new tables in one owner transaction; no schema-number relabeling is used.
|
||||
#[test]
|
||||
fn plugin_data_migration_preserves_references_and_matches_fresh_schema() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("server.db");
|
||||
prepare_retained_schema(&path, 110);
|
||||
let conn = Connection::open(&path).unwrap();
|
||||
configure_sqlite(&conn).unwrap();
|
||||
seed_migration_workspace(&conn);
|
||||
let digest = format!("sha256:{}", "a".repeat(64));
|
||||
conn.execute("INSERT INTO plugin_installations(workspace_id,installation_id,alias,package_id,last_operation) VALUES('space','installation','alias','package','install')", []).unwrap();
|
||||
conn.execute("INSERT INTO plugin_artifacts(workspace_id,installation_id,artifact_digest,package_digest,version,artifact_size,metadata_json) VALUES('space','installation',?1,?1,'1',1,'{}')", [&digest]).unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO plugin_references VALUES('space','config','config-owner','installation',?1)",
|
||||
[&digest],
|
||||
)
|
||||
.unwrap();
|
||||
let history = workspace_schema_migration_history(&conn).unwrap();
|
||||
drop(conn);
|
||||
let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
|
||||
store.with_conn(|conn| {
|
||||
assert_eq!(conn.query_row("SELECT count(*) FROM plugin_references WHERE kind='config' AND owner='config-owner'", [], |r| r.get::<_, i64>(0))?, 1);
|
||||
conn.execute("INSERT INTO plugin_references VALUES('space','data','data-owner','installation',?1)", [&digest])?;
|
||||
assert!(conn.execute("UPDATE plugin_installations SET uninstalled=1 WHERE workspace_id='space'", []).is_err());
|
||||
assert!(conn.execute("INSERT INTO plugin_references VALUES('space','unknown','bad-owner','installation',?1)", [&digest]).is_err());
|
||||
assert_eq!(&workspace_schema_migration_history(conn)?[..history.len()], history.as_slice());
|
||||
let fresh = SqliteWorkspaceStore::in_memory_for_test()?;
|
||||
let objects = |connection: &Connection| -> Result<Vec<(String,String)>> {
|
||||
Ok(connection.prepare("SELECT name,sql FROM sqlite_schema WHERE sql IS NOT NULL AND (name LIKE 'plugin_data_%' OR name='plugin_references') ORDER BY name")?.query_map([], |r| Ok((r.get(0)?,r.get(1)?)))?.collect::<rusqlite::Result<Vec<_>>>()?)
|
||||
};
|
||||
let expected = fresh.with_conn(objects)?;
|
||||
assert_eq!(objects(conn)?, expected);
|
||||
assert_eq!(conn.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |r| r.get::<_,i64>(0))?, 0);
|
||||
Ok(())
|
||||
}).unwrap();
|
||||
drop(store);
|
||||
SqliteWorkspaceStore::open_for_test(&path).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn plugin_data_migration_late_failure_rolls_back_catalog_rebuild_and_history() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("server.db");
|
||||
prepare_retained_schema(&path, 110);
|
||||
let conn = Connection::open(&path).unwrap();
|
||||
configure_sqlite(&conn).unwrap();
|
||||
conn.execute_batch("CREATE TABLE plugin_data_receipts(unrelated INTEGER)")
|
||||
.unwrap();
|
||||
let history = workspace_schema_migration_history(&conn).unwrap();
|
||||
let schema = |conn: &Connection| {
|
||||
conn.query_row("SELECT group_concat(sql,char(10)) FROM (SELECT sql FROM sqlite_schema WHERE sql IS NOT NULL ORDER BY name)", [], |r| r.get::<_,String>(0)).unwrap()
|
||||
};
|
||||
let before = schema(&conn);
|
||||
assert!(migrate_plugin_data_v110_to_v111(&conn).is_err());
|
||||
assert_eq!(schema(&conn), before);
|
||||
assert_eq!(workspace_schema_migration_history(&conn).unwrap(), history);
|
||||
conn.execute_batch("DROP TABLE plugin_data_receipts")
|
||||
.unwrap();
|
||||
migrate_plugin_data_v110_to_v111(&conn).unwrap();
|
||||
assert_eq!(current_schema_version(&conn).unwrap(), 111);
|
||||
assert!(migrate_plugin_data_v110_to_v111(&conn).is_err());
|
||||
}
|
||||
@@ -23,6 +23,7 @@ pub struct AdmittedPluginOperation {
|
||||
pub plan: StoredPluginPlan,
|
||||
pub pin: WorkerPluginPin,
|
||||
pub binding: Option<PluginConnectionBinding>,
|
||||
pub data_collections: Vec<PluginDataSchemaMetadata>,
|
||||
}
|
||||
|
||||
/// `execution` is derived by the authenticated Host boundary from current Worker
|
||||
@@ -131,7 +132,12 @@ pub fn authorize(
|
||||
} else {
|
||||
None
|
||||
};
|
||||
Ok(AdmittedPluginOperation { plan, pin, binding })
|
||||
Ok(AdmittedPluginOperation {
|
||||
plan,
|
||||
pin,
|
||||
binding,
|
||||
data_collections: artifact.data_collections,
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn live_binding(
|
||||
|
||||
@@ -12,17 +12,17 @@ fn plugin_receipt_tail_migrates_true_predecessor_reopens_and_matches_fresh_schem
|
||||
assert!(!table_exists(&conn, "plugin_invocations").unwrap());
|
||||
let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap();
|
||||
assert_eq!(plan.current_schema_version, 108);
|
||||
assert_eq!(plan.migrations.iter().map(|step| step.version).collect::<Vec<_>>(), [109, 110]);
|
||||
assert_eq!(plan.migrations.iter().map(|step| step.version).collect::<Vec<_>>(), [109, 110, 111]);
|
||||
assert_eq!(current_schema_version(&conn).unwrap(), 108);
|
||||
drop(conn);
|
||||
let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
|
||||
store.with_conn(|conn| {
|
||||
crate::plugin_host::verify_receipt_schema(conn)?;
|
||||
assert_eq!(current_schema_version(conn)?, 110);
|
||||
assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION);
|
||||
let upgraded = workspace_schema_migration_history(conn)?;
|
||||
assert_eq!(&upgraded[..history.len()], history.as_slice());
|
||||
assert_eq!(upgraded.iter().find(|(version, _)| *version == 109).unwrap().1, "Durable Plugin Host invocations and HTTPS effect receipts");
|
||||
assert_eq!(upgraded.last().unwrap().1, "Worker Plugin launch selection associations");
|
||||
assert_eq!(upgraded.iter().find(|(version, _)| *version == 110).unwrap().1, "Worker Plugin launch selection associations");
|
||||
assert_eq!(conn.query_row("SELECT count(*) FROM workspaces WHERE workspace_id='space'", [], |row| row.get::<_, i64>(0))?, 1);
|
||||
assert_eq!(conn.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |row| row.get::<_, i64>(0))?, 0);
|
||||
Ok(())
|
||||
|
||||
@@ -35,6 +35,10 @@ fn plugin_plan_migration_marker_failure_preserves_catalog_then_retries_and_match
|
||||
"Durable Plugin Host invocations and HTTPS effect receipts".into(),
|
||||
));
|
||||
expected.push((110, "Worker Plugin launch selection associations".into()));
|
||||
expected.push((
|
||||
111,
|
||||
"Typed Plugin Ticket extension data and atomic mutation receipts".into(),
|
||||
));
|
||||
assert_eq!(workspace_schema_migration_history(conn)?, expected);
|
||||
subject_assignment_migration::verify(conn)?;
|
||||
crate::plugin_plan::verify_plan_schema(conn)?;
|
||||
|
||||
@@ -227,7 +227,10 @@ pub fn lookup(conn: &Connection, workspace: &str, invocation_id: &str) -> Result
|
||||
|
||||
// Never trust a passed record's mutable flags/deadline or accept a fabricated
|
||||
// plan/execution/consumer. Records may be stale only in monotonic terminal flags.
|
||||
fn verify_record(conn: &Connection, supplied: &InvocationRecord) -> Result<InvocationRecord> {
|
||||
pub(crate) fn verify_record(
|
||||
conn: &Connection,
|
||||
supplied: &InvocationRecord,
|
||||
) -> Result<InvocationRecord> {
|
||||
let saved = lookup(conn, &supplied.workspace_id, &supplied.invocation_id)?;
|
||||
if supplied.plan_id != saved.plan_id
|
||||
|| supplied.installation_id != saved.installation_id
|
||||
@@ -334,7 +337,11 @@ pub fn effect_admit(
|
||||
tx.execute(
|
||||
"INSERT INTO plugin_effect_receipts(workspace_id,invocation_id,effect_key,fingerprint,admitted_at_ms,state,error_json) VALUES(?1,?2,?3,?4,?5,'pending',?6)",
|
||||
params![record.workspace_id, record.invocation_id, effect_key, fingerprint, now_ms, encode(&unknown)?],
|
||||
)?;
|
||||
).map_err(|error| match &error {
|
||||
rusqlite::Error::SqliteFailure(_, Some(message)) if message == "Plugin operation key belongs to data" =>
|
||||
Error::from(host_error(PluginHostErrorCode::Conflict, PluginEffectObservation::Indeterminate)),
|
||||
_ => Error::from(error),
|
||||
})?;
|
||||
Ok(EffectAdmission::Dispatch)
|
||||
}
|
||||
|
||||
@@ -441,8 +448,9 @@ pub fn cancel(tx: &Transaction<'_>, workspace: &str, invocation_id: &str) -> Res
|
||||
}
|
||||
|
||||
/// Expire matching invocations and erase response/error payloads, never identity,
|
||||
/// fingerprints or keys. Returns the number of receipts newly tombstoned. Even a
|
||||
/// subsequent backwards clock cannot resurrect an explicitly expired invocation.
|
||||
/// fingerprints or keys. Returns the number of HTTPS receipts newly tombstoned;
|
||||
/// extension-data payload expiry follows the same invocation flag atomically.
|
||||
/// A backwards clock cannot resurrect an explicitly expired invocation.
|
||||
/// Tombstones survive until the parent's Workspace deletion cascade.
|
||||
pub fn prune(tx: &Transaction<'_>, workspace: &str, now_ms: i64) -> Result<usize> {
|
||||
if !valid_identity(workspace, 512) || now_ms < 0 {
|
||||
|
||||
@@ -140,6 +140,7 @@ fn make_plan(
|
||||
}],
|
||||
connections: vec![],
|
||||
requested_grants: vec![],
|
||||
data_collections: vec![],
|
||||
config_schema: PluginSchemaContribution {
|
||||
identity: "schema".into(),
|
||||
version: "1".into(),
|
||||
|
||||
@@ -46,6 +46,12 @@ pub struct PluginEnablement {
|
||||
pub artifact_digest: String,
|
||||
pub tools: Vec<String>,
|
||||
pub logging: bool,
|
||||
#[serde(default)]
|
||||
pub ticket_read: bool,
|
||||
#[serde(default)]
|
||||
pub ticket_data_read: Vec<manifest::PluginTicketDataGrant>,
|
||||
#[serde(default)]
|
||||
pub ticket_data_write: Vec<manifest::PluginTicketDataGrant>,
|
||||
pub connections: BTreeMap<String, String>,
|
||||
pub config: serde_json::Value,
|
||||
}
|
||||
@@ -164,7 +170,7 @@ pub(crate) fn evaluate_catalog_config(
|
||||
// Each alias has its own closed schema. Reusing an artifact with distinct
|
||||
// bindings is intentional and does not manufacture another schema authority.
|
||||
let enabled_fields = artifacts.iter().map(|(alias, artifact)| {
|
||||
format!("{} = {{ installation_id = String; expected_last_operation = String; package_digest = String; artifact_digest = String; tools = [...String]; logging = Bool default false; connections = {{ ...String }} default {{}}; config = ({}); }};", alias, artifact.config_schema.dcdl)
|
||||
format!("{} = {{ installation_id = String; expected_last_operation = String; package_digest = String; artifact_digest = String; tools = [...String]; logging = Bool default false; ticket_read = Bool default false; ticket_data_read = [...{{ collection = String; schema_id = String; schema_version = String; }}] default []; ticket_data_write = [...{{ collection = String; schema_id = String; schema_version = String; }}] default []; connections = {{ ...String }} default {{}}; config = ({}); }};", alias, artifact.config_schema.dcdl)
|
||||
}).collect::<Vec<_>>().join("\n");
|
||||
let enabled_schema = if enabled_fields.is_empty() {
|
||||
"{} default {}".to_string()
|
||||
@@ -238,6 +244,7 @@ fn validate_enablement(
|
||||
| manifest::plugin::PLUGIN_COMPONENT_HTTPS_TOOL_WORLD
|
||||
| manifest::plugin::PLUGIN_COMPONENT_LOGGED_TOOL_WORLD
|
||||
| manifest::plugin::PLUGIN_COMPONENT_HTTPS_LOGGED_TOOL_WORLD
|
||||
| "yoi:plugin/ticket-tool-plugin@2.1.0"
|
||||
)
|
||||
|| artifact.required_imports.iter().any(|i| {
|
||||
!matches!(
|
||||
@@ -246,6 +253,9 @@ fn validate_enablement(
|
||||
| "yoi:host/connections@2.0.0"
|
||||
| "yoi:host/http@2.0.0"
|
||||
| "yoi:host/logging@2.0.0"
|
||||
| "yoi:host/objects@2.1.0"
|
||||
| "yoi:host/ticket-read@2.1.0"
|
||||
| "yoi:host/extension-data@2.1.0"
|
||||
)
|
||||
})
|
||||
{
|
||||
@@ -266,6 +276,39 @@ fn validate_enablement(
|
||||
{
|
||||
return Err(config_error("logging exceeds package request"));
|
||||
}
|
||||
if enabled.ticket_read
|
||||
&& !artifact
|
||||
.requested_grants
|
||||
.contains(&PluginGrantMetadata::TicketRead)
|
||||
{
|
||||
return Err(config_error("Ticket read exceeds package request"));
|
||||
}
|
||||
for (write, grants) in [
|
||||
(false, &enabled.ticket_data_read),
|
||||
(true, &enabled.ticket_data_write),
|
||||
] {
|
||||
if grants.iter().collect::<BTreeSet<_>>().len() != grants.len() {
|
||||
return Err(config_error("duplicate Ticket data grant"));
|
||||
}
|
||||
for grant in grants {
|
||||
let metadata = if write {
|
||||
PluginGrantMetadata::TicketDataWrite {
|
||||
collection: grant.collection.clone(),
|
||||
schema_id: grant.schema_id.clone(),
|
||||
schema_version: grant.schema_version.clone(),
|
||||
}
|
||||
} else {
|
||||
PluginGrantMetadata::TicketDataRead {
|
||||
collection: grant.collection.clone(),
|
||||
schema_id: grant.schema_id.clone(),
|
||||
schema_version: grant.schema_version.clone(),
|
||||
}
|
||||
};
|
||||
if !artifact.requested_grants.contains(&metadata) {
|
||||
return Err(config_error("Ticket data grant exceeds package request"));
|
||||
}
|
||||
}
|
||||
}
|
||||
for (slot, binding) in &enabled.connections {
|
||||
let requested = artifact
|
||||
.connections
|
||||
@@ -390,6 +433,15 @@ fn project_profile_plugins(
|
||||
.iter()
|
||||
.any(|tool| !enabled.tools.contains(tool))
|
||||
|| (selection.logging && !enabled.logging)
|
||||
|| (selection.ticket_read && !enabled.ticket_read)
|
||||
|| selection
|
||||
.ticket_data_read
|
||||
.iter()
|
||||
.any(|grant| !enabled.ticket_data_read.contains(grant))
|
||||
|| selection
|
||||
.ticket_data_write
|
||||
.iter()
|
||||
.any(|grant| !enabled.ticket_data_write.contains(grant))
|
||||
|| selection
|
||||
.https_slots
|
||||
.iter()
|
||||
@@ -412,6 +464,23 @@ fn project_profile_plugins(
|
||||
if selection.logging {
|
||||
grants.push(PluginGrant::Logging);
|
||||
}
|
||||
if selection.ticket_read {
|
||||
grants.push(PluginGrant::TicketRead);
|
||||
}
|
||||
for grant in &selection.ticket_data_read {
|
||||
grants.push(PluginGrant::TicketDataRead {
|
||||
collection: grant.collection.clone(),
|
||||
schema_id: grant.schema_id.clone(),
|
||||
schema_version: grant.schema_version.clone(),
|
||||
});
|
||||
}
|
||||
for grant in &selection.ticket_data_write {
|
||||
grants.push(PluginGrant::TicketDataWrite {
|
||||
collection: grant.collection.clone(),
|
||||
schema_id: grant.schema_id.clone(),
|
||||
schema_version: grant.schema_version.clone(),
|
||||
});
|
||||
}
|
||||
let mut bindings = Vec::new();
|
||||
for slot in &selection.https_slots {
|
||||
let binding_id = &enabled.connections[slot];
|
||||
|
||||
@@ -120,6 +120,10 @@ mod schema_convergence {
|
||||
version: 110,
|
||||
name: "Worker Plugin launch selection associations".into(),
|
||||
},
|
||||
WorkspaceSchemaMigrationStep {
|
||||
version: 111,
|
||||
name: "Typed Plugin Ticket extension data and atomic mutation receipts".into(),
|
||||
},
|
||||
]
|
||||
);
|
||||
assert_eq!(
|
||||
@@ -141,6 +145,7 @@ mod schema_convergence {
|
||||
expected.push((108, "Subject Ticket responsibility and current-body notification routing".into()));
|
||||
expected.push((109, "Durable Plugin Host invocations and HTTPS effect receipts".into()));
|
||||
expected.push((110, "Worker Plugin launch selection associations".into()));
|
||||
expected.push((111, "Typed Plugin Ticket extension data and atomic mutation receipts".into()));
|
||||
assert_eq!(workspace_schema_migration_history(conn)?, expected);
|
||||
for table in ["passkey_credentials", "passkey_metadata", "passkey_addition_receipts", "backend_jobs", "backend_job_attempts"] {
|
||||
assert_eq!(state_contract_schema_signature(conn, table), fresh.with_conn(|c| Ok(state_contract_schema_signature(c, table)))?, "{table}");
|
||||
|
||||
@@ -13806,6 +13806,33 @@ impl server_api::ServerApi for ServerApiContractService {
|
||||
plugin::uninstall(api, actor, workspace_id, installation_id, request).await
|
||||
}
|
||||
|
||||
async fn plugin_data_observe(
|
||||
&self,
|
||||
actor: RequestActor,
|
||||
workspace_id: String,
|
||||
installation_id: String,
|
||||
) -> std::result::Result<server_api::PluginDataPurgeState, server_api::RepositoryApiError> {
|
||||
plugin::data_observe(self.workspace_api()?, actor, workspace_id, installation_id).await
|
||||
}
|
||||
|
||||
async fn plugin_data_purge(
|
||||
&self,
|
||||
actor: RequestActor,
|
||||
workspace_id: String,
|
||||
installation_id: String,
|
||||
request: server_api::PluginDataPurgeRequest,
|
||||
) -> std::result::Result<server_api::PluginDataPurgeResponse, server_api::RepositoryApiError>
|
||||
{
|
||||
plugin::data_purge(
|
||||
self.workspace_api()?,
|
||||
actor,
|
||||
workspace_id,
|
||||
installation_id,
|
||||
request,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn plugin_operation(
|
||||
&self,
|
||||
actor: RequestActor,
|
||||
|
||||
@@ -21,6 +21,8 @@ pub(super) fn router(service: Arc<ServerApiContractService>) -> Router {
|
||||
.merge(server_api_axum::plugin_artifact_add(service.clone()))
|
||||
.merge(server_api_axum::plugin_revoke(service.clone()))
|
||||
.merge(server_api_axum::plugin_uninstall(service.clone()))
|
||||
.merge(server_api_axum::plugin_data_observe(service.clone()))
|
||||
.merge(server_api_axum::plugin_data_purge(service.clone()))
|
||||
.merge(server_api_axum::plugin_operation(service))
|
||||
.layer(middleware::from_fn(normalize_rejections))
|
||||
}
|
||||
@@ -225,3 +227,85 @@ pub(super) async fn operation(
|
||||
.await
|
||||
.map_err(failure)
|
||||
}
|
||||
|
||||
fn data_failure(error: Error) -> RepositoryApiError {
|
||||
if let Error::PluginHost(error) = error {
|
||||
let (status, code) = match error.code {
|
||||
PluginHostErrorCode::NotFound => (404, "not_found"),
|
||||
PluginHostErrorCode::Denied => (403, "access_denied"),
|
||||
PluginHostErrorCode::Conflict => (409, "plugin_data_conflict"),
|
||||
PluginHostErrorCode::InvalidInput => (400, "plugin_invalid"),
|
||||
_ => (503, "plugin_data_outcome_unknown"),
|
||||
};
|
||||
RepositoryApiError::new(status, code, "Plugin data operation", vec![])
|
||||
} else {
|
||||
failure(error)
|
||||
}
|
||||
}
|
||||
fn purge_state(state: crate::plugin_data::PurgeSnapshot) -> PluginDataPurgeState {
|
||||
PluginDataPurgeState {
|
||||
installation_last_operation: state.installation_last_operation,
|
||||
cells_digest: state.cells_digest,
|
||||
cell_count: state.cell_count,
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) async fn data_observe(
|
||||
api: &WorkspaceApi,
|
||||
actor: RequestActor,
|
||||
workspace: String,
|
||||
installation: String,
|
||||
) -> CatalogResult<PluginDataPurgeState> {
|
||||
// Observing a purge confirmation is an administrator operation. It is only
|
||||
// available after all execution artifacts have been irrevocably revoked.
|
||||
authorize(api, &workspace, &actor, true).await?;
|
||||
let store = api.config_store.clone();
|
||||
store
|
||||
.db_executor()
|
||||
.run(move || {
|
||||
store.with_immediate_transaction(|tx| {
|
||||
crate::plugin_data::purge_snapshot(tx, &workspace, &installation)
|
||||
})
|
||||
})
|
||||
.await
|
||||
.map(purge_state)
|
||||
.map_err(data_failure)
|
||||
}
|
||||
|
||||
pub(super) async fn data_purge(
|
||||
api: &WorkspaceApi,
|
||||
actor: RequestActor,
|
||||
workspace: String,
|
||||
installation: String,
|
||||
request: PluginDataPurgeRequest,
|
||||
) -> CatalogResult<PluginDataPurgeResponse> {
|
||||
// Authenticate again even when this will be a receipt replay. Management
|
||||
// credentials are never borrowed by guest invocations or used to run them.
|
||||
authorize(api, &workspace, &actor, true).await?;
|
||||
let actor = actor_identity(&actor);
|
||||
let store = api.config_store.clone();
|
||||
store
|
||||
.db_executor()
|
||||
.run(move || {
|
||||
store.with_immediate_transaction(|tx| {
|
||||
crate::plugin_data::purge(
|
||||
tx,
|
||||
&workspace,
|
||||
&installation,
|
||||
&actor,
|
||||
&request.operation_key,
|
||||
&crate::plugin_data::PurgeSnapshot {
|
||||
installation_last_operation: request.expected.installation_last_operation,
|
||||
cells_digest: request.expected.cells_digest,
|
||||
cell_count: request.expected.cell_count,
|
||||
},
|
||||
)
|
||||
})
|
||||
})
|
||||
.await
|
||||
.map(|receipt| PluginDataPurgeResponse {
|
||||
purge_id: receipt.purge_id,
|
||||
observed: purge_state(receipt.observed),
|
||||
})
|
||||
.map_err(data_failure)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,331 @@
|
||||
//! Narrow Ticket owner adapter. Resource selectors never replace current
|
||||
//! Subject/delegation authorization; values are projected explicitly and never
|
||||
//! serialized from the general Ticket detail (which contains thread/admin data).
|
||||
use super::*;
|
||||
use crate::plugin_catalog::{PluginReference, PluginReferenceKind};
|
||||
use crate::plugin_data as store;
|
||||
use server_api::authorization::{OperationAuthorizationRequest, PermissionResource};
|
||||
|
||||
fn not_found() -> PluginHostError {
|
||||
policy::map_error(policy::failure(PluginHostErrorCode::NotFound))
|
||||
}
|
||||
fn invalid() -> PluginHostError {
|
||||
policy::map_error(policy::failure(PluginHostErrorCode::InvalidInput))
|
||||
}
|
||||
|
||||
fn collection_granted(
|
||||
admitted: &policy::AdmittedPluginOperation,
|
||||
collection: &PluginDataCollection,
|
||||
write: bool,
|
||||
) -> bool {
|
||||
admitted
|
||||
.pin
|
||||
.projection()
|
||||
.grants
|
||||
.iter()
|
||||
.any(|grant| match grant {
|
||||
crate::plugin_plan::PluginGrant::TicketDataRead {
|
||||
collection: c,
|
||||
schema_id,
|
||||
schema_version,
|
||||
} if !write => {
|
||||
c == &collection.collection
|
||||
&& schema_id == &collection.schema_id
|
||||
&& schema_version == &collection.schema_version
|
||||
}
|
||||
crate::plugin_plan::PluginGrant::TicketDataWrite {
|
||||
collection: c,
|
||||
schema_id,
|
||||
schema_version,
|
||||
} if write => {
|
||||
c == &collection.collection
|
||||
&& schema_id == &collection.schema_id
|
||||
&& schema_version == &collection.schema_version
|
||||
}
|
||||
_ => false,
|
||||
})
|
||||
}
|
||||
|
||||
fn access_granted(admitted: &policy::AdmittedPluginOperation, access: PluginTicketAccess) -> bool {
|
||||
admitted
|
||||
.pin
|
||||
.projection()
|
||||
.grants
|
||||
.iter()
|
||||
.any(|grant| match (access, grant) {
|
||||
(
|
||||
PluginTicketAccess::Read,
|
||||
crate::plugin_plan::PluginGrant::TicketRead
|
||||
| crate::plugin_plan::PluginGrant::TicketDataRead { .. },
|
||||
) => true,
|
||||
(
|
||||
PluginTicketAccess::ExtensionWrite,
|
||||
crate::plugin_plan::PluginGrant::TicketDataWrite { .. },
|
||||
) => true,
|
||||
_ => false,
|
||||
})
|
||||
}
|
||||
|
||||
fn authorize_ticket(
|
||||
api: &WorkspaceApi,
|
||||
context: &ServerRequestContext,
|
||||
workspace: &str,
|
||||
reference: &str,
|
||||
access: PluginTicketAccess,
|
||||
) -> crate::Result<String> {
|
||||
let hidden = || policy::failure(PluginHostErrorCode::NotFound);
|
||||
if reference.is_empty() || reference.len() > 512 || reference.chars().any(char::is_control) {
|
||||
return Err(hidden());
|
||||
}
|
||||
let worker = subject_authorization::source_from_context(context).map_err(|e| e.error)?;
|
||||
let ticket_id = api
|
||||
.store
|
||||
.resolve_resource_reference(workspace, WorkspaceResourceKind::Ticket, reference)?
|
||||
.ok_or_else(hidden)?;
|
||||
subject_authorization::authorize(
|
||||
api,
|
||||
&worker,
|
||||
&OperationAuthorizationRequest {
|
||||
operation: match access {
|
||||
PluginTicketAccess::Read => "ticket.read",
|
||||
PluginTicketAccess::ExtensionWrite => "ticket.extension.write",
|
||||
}
|
||||
.into(),
|
||||
resource: PermissionResource::Ticket {
|
||||
workspace_id: workspace.into(),
|
||||
ticket_id: ticket_id.clone(),
|
||||
},
|
||||
},
|
||||
)
|
||||
.map_err(|e| match e.error {
|
||||
Error::WorkspacePermissionDenied(_) => hidden(),
|
||||
other => other,
|
||||
})?;
|
||||
Ok(ticket_id)
|
||||
}
|
||||
|
||||
fn read_ticket(
|
||||
tx: &rusqlite::Transaction<'_>,
|
||||
workspace: &str,
|
||||
ticket: &str,
|
||||
) -> crate::Result<PluginTicketSnapshot> {
|
||||
let (title, status, body): (String, String, String) = tx.query_row(
|
||||
"SELECT t.title,t.status,e.body FROM typed_tickets t JOIN typed_ticket_events e ON e.workspace_id=t.workspace_id AND e.ticket_id=t.ticket_id AND e.event_index=-1 WHERE t.workspace_id=?1 AND t.ticket_id=?2 AND EXISTS(SELECT 1 FROM workspaces WHERE workspace_id=t.workspace_id AND state='active')",
|
||||
rusqlite::params![workspace,ticket], |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?)),
|
||||
).optional()?.ok_or_else(|| policy::failure(PluginHostErrorCode::NotFound))?;
|
||||
let content_digest = ticket::sqlite_ticket_content_digest(tx, workspace, ticket)?
|
||||
.ok_or_else(|| policy::failure(PluginHostErrorCode::NotFound))?;
|
||||
let truncated = body.len() > PLUGIN_TICKET_BODY_LIMIT;
|
||||
let mut end = body.len().min(PLUGIN_TICKET_BODY_LIMIT);
|
||||
while !body.is_char_boundary(end) {
|
||||
end -= 1;
|
||||
}
|
||||
let case_state = match status.as_str() {
|
||||
"open" => PluginTicketCaseState::Open,
|
||||
"closed" => PluginTicketCaseState::Closed,
|
||||
_ => return Err(policy::failure(PluginHostErrorCode::Unavailable)),
|
||||
};
|
||||
Ok(PluginTicketSnapshot {
|
||||
title,
|
||||
case_state,
|
||||
content_digest,
|
||||
body: body[..end].into(),
|
||||
truncated,
|
||||
})
|
||||
}
|
||||
|
||||
fn reference(
|
||||
tx: &rusqlite::Transaction<'_>,
|
||||
workspace: &str,
|
||||
admitted: &policy::AdmittedPluginOperation,
|
||||
) -> crate::Result<PluginReference> {
|
||||
let expected_last_operation = tx.query_row(
|
||||
"SELECT last_operation FROM plugin_installations WHERE workspace_id=?1 AND installation_id=?2 AND uninstalled=0",
|
||||
rusqlite::params![workspace,admitted.pin.installation_id()], |row| row.get(0),
|
||||
).optional()?.ok_or_else(|| policy::failure(PluginHostErrorCode::Denied))?;
|
||||
Ok(PluginReference {
|
||||
kind: PluginReferenceKind::Data,
|
||||
owner: String::new(),
|
||||
installation_id: admitted.pin.installation_id().into(),
|
||||
expected_last_operation,
|
||||
package_digest: admitted.pin.package_digest().into(),
|
||||
artifact_digest: admitted.pin.artifact_digest().into(),
|
||||
})
|
||||
}
|
||||
|
||||
fn schema<'a>(
|
||||
admitted: &'a policy::AdmittedPluginOperation,
|
||||
collection: &PluginDataCollection,
|
||||
) -> crate::Result<&'a PluginDataSchemaMetadata> {
|
||||
admitted
|
||||
.data_collections
|
||||
.iter()
|
||||
.find(|declaration| {
|
||||
declaration.collection == collection.collection
|
||||
&& declaration.schema_id == collection.schema_id
|
||||
&& declaration.schema_version == collection.schema_version
|
||||
})
|
||||
.ok_or_else(|| policy::failure(PluginHostErrorCode::Denied))
|
||||
}
|
||||
fn validate_value(
|
||||
declaration: &PluginDataSchemaMetadata,
|
||||
mutation: &PluginDataMutation,
|
||||
) -> crate::Result<()> {
|
||||
if let PluginDataMutation::Put { value_json, .. } = mutation {
|
||||
if value_json.len() > PLUGIN_DATA_VALUE_LIMIT {
|
||||
return Err(policy::failure(PluginHostErrorCode::QuotaExceeded));
|
||||
}
|
||||
let value: serde_json::Value = serde_json::from_str(value_json)
|
||||
.map_err(|_| policy::failure(PluginHostErrorCode::InvalidInput))?;
|
||||
manifest::plugin::schema::validate_data_value(&declaration.schema, &value)
|
||||
.map_err(|_| policy::failure(PluginHostErrorCode::InvalidInput))?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
enum Action {
|
||||
Resolve(PluginTicketAccess),
|
||||
Read,
|
||||
Get(PluginDataAddress),
|
||||
Mutate(PluginDataAddress, String, PluginDataMutation),
|
||||
List(PluginDataCollection, Option<String>, u32),
|
||||
Lookup(PluginDataAddress, String),
|
||||
}
|
||||
impl Action {
|
||||
fn access(&self) -> PluginTicketAccess {
|
||||
match self {
|
||||
Self::Resolve(access) => *access,
|
||||
Self::Mutate(..) | Self::Lookup(..) => PluginTicketAccess::ExtensionWrite,
|
||||
_ => PluginTicketAccess::Read,
|
||||
}
|
||||
}
|
||||
fn collection(&self) -> Option<&PluginDataCollection> {
|
||||
match self {
|
||||
Self::Get(a) | Self::Mutate(a, ..) | Self::Lookup(a, ..) => Some(&a.collection),
|
||||
Self::List(c, ..) => Some(c),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) async fn execute(
|
||||
api: Arc<WorkspaceApi>,
|
||||
context: ServerRequestContext,
|
||||
workspace: String,
|
||||
request: PluginHostRequest,
|
||||
) -> std::result::Result<PluginHostResponse, PluginHostError> {
|
||||
let (invocation_id, selector, action) = match request {
|
||||
PluginHostRequest::ResolveTicket {
|
||||
invocation_id,
|
||||
ticket_ref,
|
||||
access,
|
||||
} => (invocation_id, ticket_ref, Action::Resolve(access)),
|
||||
PluginHostRequest::ReadTicket {
|
||||
invocation_id,
|
||||
ticket_id,
|
||||
} => (invocation_id, ticket_id, Action::Read),
|
||||
PluginHostRequest::DataGet {
|
||||
invocation_id,
|
||||
ticket_id,
|
||||
address,
|
||||
} => (invocation_id, ticket_id, Action::Get(address)),
|
||||
PluginHostRequest::DataMutate {
|
||||
invocation_id,
|
||||
ticket_id,
|
||||
address,
|
||||
operation_key,
|
||||
mutation,
|
||||
} => (
|
||||
invocation_id,
|
||||
ticket_id,
|
||||
Action::Mutate(address, operation_key, mutation),
|
||||
),
|
||||
PluginHostRequest::DataList {
|
||||
invocation_id,
|
||||
ticket_id,
|
||||
collection,
|
||||
cursor,
|
||||
limit,
|
||||
} => (
|
||||
invocation_id,
|
||||
ticket_id,
|
||||
Action::List(collection, cursor, limit),
|
||||
),
|
||||
PluginHostRequest::DataLookupMutation {
|
||||
invocation_id,
|
||||
ticket_id,
|
||||
address,
|
||||
operation_key,
|
||||
} => (
|
||||
invocation_id,
|
||||
ticket_id,
|
||||
Action::Lookup(address, operation_key),
|
||||
),
|
||||
_ => return Err(invalid()),
|
||||
};
|
||||
// Lookup after expiry must remain distinguishable from NotRecorded. It still
|
||||
// goes through the exact same live owner checks and current Ticket authority.
|
||||
let config = api.config_store.clone();
|
||||
let w = workspace.clone();
|
||||
let id = invocation_id.clone();
|
||||
let record = config
|
||||
.db_executor()
|
||||
.run(move || config.with_conn(|conn| receipt::lookup(conn, &w, &id)))
|
||||
.await
|
||||
.map_err(policy::map_error)?;
|
||||
let admitted = check(
|
||||
api.clone(),
|
||||
context.clone(),
|
||||
workspace.clone(),
|
||||
record.plan_id.clone(),
|
||||
record.installation_id.clone(),
|
||||
record.consumer.clone(),
|
||||
PluginHostOperation::Admit,
|
||||
)
|
||||
.await?;
|
||||
if record.execution != *admitted.plan.plan.execution() {
|
||||
return Err(denied());
|
||||
}
|
||||
if !access_granted(&admitted, action.access()) {
|
||||
return Err(not_found());
|
||||
}
|
||||
let config = api.config_store.clone();
|
||||
let executor = config.db_executor();
|
||||
executor.run(move || {
|
||||
// Resolve aliases and the live Subject/delegation ceiling before any
|
||||
// Ticket content or receipt is read. The DB transaction then fences
|
||||
// catalog revoke, cancellation, parent deletion and competing writes.
|
||||
let ticket_id = authorize_ticket(&api, &context, &workspace, &selector, action.access())?;
|
||||
config.with_immediate_transaction(|tx| {
|
||||
let record = receipt::verify_record(tx, &record)?;
|
||||
let tools = match &record.consumer { PluginConsumer::ToolCall { tool_name, .. } => vec![tool_name.clone()], _ => return Err(policy::failure(PluginHostErrorCode::Unsupported)) };
|
||||
let admitted = policy::authorize(tx, &workspace, &record.execution, &record.plan_id, &record.installation_id, &tools, &PluginHostOperation::Admit)?;
|
||||
if record.cancelled { return Err(policy::failure(PluginHostErrorCode::Cancelled)); }
|
||||
if !matches!(action, Action::Lookup(..)) && (record.expired || record.expires_at_ms <= now_ms()) { return Err(policy::failure(PluginHostErrorCode::ReceiptExpired)); }
|
||||
if !tx.query_row("SELECT EXISTS(SELECT 1 FROM typed_tickets t JOIN workspaces w USING(workspace_id) WHERE t.workspace_id=?1 AND t.ticket_id=?2 AND w.state='active')",rusqlite::params![workspace,ticket_id],|row|row.get::<_,bool>(0))? { return Err(policy::failure(PluginHostErrorCode::NotFound)); }
|
||||
if let Some(collection) = action.collection() {
|
||||
let write = matches!(action, Action::Mutate(..) | Action::Lookup(..));
|
||||
if !collection_granted(&admitted, collection, write) { return Err(policy::failure(PluginHostErrorCode::Denied)); }
|
||||
schema(&admitted, collection)?;
|
||||
}
|
||||
match action {
|
||||
Action::Resolve(_) => Ok(PluginHostResponse::TicketResolved { ticket_id }),
|
||||
Action::Read => {
|
||||
if !admitted.pin.projection().grants.contains(&crate::plugin_plan::PluginGrant::TicketRead) { return Err(policy::failure(PluginHostErrorCode::Denied)); }
|
||||
Ok(PluginHostResponse::TicketRead { snapshot: read_ticket(tx, &workspace, &ticket_id)? })
|
||||
}
|
||||
Action::Get(address) => Ok(PluginHostResponse::DataCell { cell: store::get(tx, &store::namespace(&record, &ticket_id, &address), &address.key)? }),
|
||||
Action::Mutate(address, operation_key, mutation) => {
|
||||
validate_value(schema(&admitted, &address.collection)?, &mutation)?;
|
||||
let receipt = store::mutate(tx, &record, &store::namespace(&record, &ticket_id, &address), &address.key, &operation_key, &mutation, &reference(tx, &workspace, &admitted)?, now_ms())?;
|
||||
Ok(PluginHostResponse::DataMutated { receipt })
|
||||
}
|
||||
Action::List(collection, cursor, limit) => {
|
||||
let namespace = store::Namespace { workspace_id: workspace, installation_id: record.installation_id, ticket_id, collection };
|
||||
Ok(PluginHostResponse::DataListed { page: store::list(tx, &namespace, cursor.as_deref(), limit, now_ms())? })
|
||||
}
|
||||
Action::Lookup(address, operation_key) => Ok(PluginHostResponse::DataMutationLookup { outcome: store::lookup_mutation(tx, &record, &store::namespace(&record, &ticket_id, &address), &address.key, &operation_key, now_ms())? }),
|
||||
}
|
||||
})
|
||||
}).await.map_err(policy::map_error)
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -5,6 +5,8 @@ use super::*;
|
||||
use crate::{plugin_host as policy, plugin_https as https, plugin_receipt as receipt};
|
||||
use rusqlite::OptionalExtension;
|
||||
use server_api::*;
|
||||
#[path = "server_plugin_data.rs"]
|
||||
mod data;
|
||||
|
||||
impl WorkspaceApi {
|
||||
/// Host plan creation after the canonical reservation and grant/Job binding,
|
||||
@@ -392,6 +394,14 @@ async fn execute_with_transport(
|
||||
// Human management and a Runtime-only (no Worker) source both fail closed.
|
||||
subject_authorization::source_from_context(&context).map_err(|_| denied())?;
|
||||
match request {
|
||||
request @ (PluginHostRequest::ResolveTicket { .. }
|
||||
| PluginHostRequest::ReadTicket { .. }
|
||||
| PluginHostRequest::DataGet { .. }
|
||||
| PluginHostRequest::DataMutate { .. }
|
||||
| PluginHostRequest::DataList { .. }
|
||||
| PluginHostRequest::DataLookupMutation { .. }) => {
|
||||
data::execute(api, context, workspace, request).await
|
||||
}
|
||||
PluginHostRequest::Prepare {
|
||||
plan_id,
|
||||
installation_id,
|
||||
|
||||
@@ -4,6 +4,9 @@
|
||||
// proof verification, current Subject permissions and all Host checks are real.
|
||||
// Run with `cargo test -p yoi-workspace-server --lib plugin_host_tests`.
|
||||
// The shared TLS fixture requires openssl; tests never mutate process environment.
|
||||
#[path = "server_plugin_data_tests.rs"]
|
||||
mod data;
|
||||
|
||||
use super::*;
|
||||
use crate::plugin_binding::PluginConnectionBinding;
|
||||
use crate::plugin_https::{
|
||||
@@ -48,17 +51,6 @@ struct Fixture {
|
||||
|
||||
impl Fixture {
|
||||
async fn new(binding: Option<PluginConnectionBinding>) -> Self {
|
||||
use manifest::plugin::*;
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let mut api = test_api(directory.path()).await;
|
||||
register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await;
|
||||
let runtime = WorkdirlessFixtureRuntime::default();
|
||||
api.runtime.register_or_replace(runtime.clone());
|
||||
let subject = open_subjektiv_store(&api)
|
||||
.unwrap()
|
||||
.create_subject(crate::subjektiv::SubjectRole::new("plugin-host-integration").unwrap())
|
||||
.unwrap()
|
||||
.id;
|
||||
let https = binding.is_some();
|
||||
let (manifest, component, tool) = if https {
|
||||
(include_str!("../../../resources/plugin/templates/rust-component-https/plugin.toml")
|
||||
@@ -73,6 +65,27 @@ impl Fixture {
|
||||
"example_echo",
|
||||
)
|
||||
};
|
||||
Self::with_package(binding, manifest, component, tool, None).await
|
||||
}
|
||||
|
||||
async fn with_package(
|
||||
binding: Option<PluginConnectionBinding>,
|
||||
manifest: String,
|
||||
component: &[u8],
|
||||
tool: &'static str,
|
||||
ticket_grants: Option<Value>,
|
||||
) -> Self {
|
||||
use manifest::plugin::*;
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let mut api = test_api(directory.path()).await;
|
||||
register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await;
|
||||
let runtime = WorkdirlessFixtureRuntime::default();
|
||||
api.runtime.register_or_replace(runtime.clone());
|
||||
let subject = open_subjektiv_store(&api)
|
||||
.unwrap()
|
||||
.create_subject(crate::subjektiv::SubjectRole::new("plugin-host-integration").unwrap())
|
||||
.unwrap()
|
||||
.id;
|
||||
let package_dir = tempfile::tempdir().unwrap();
|
||||
fs::write(package_dir.path().join("plugin.toml"), manifest).unwrap();
|
||||
fs::write(package_dir.path().join("plugin.component.wasm"), component).unwrap();
|
||||
@@ -105,7 +118,13 @@ impl Fixture {
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
commit_config(&api, &installed, tool, binding.as_ref());
|
||||
commit_config_with_ticket_grants(
|
||||
&api,
|
||||
&installed,
|
||||
tool,
|
||||
binding.as_ref(),
|
||||
ticket_grants.as_ref(),
|
||||
);
|
||||
let Json(created) = create_workspace_worker(
|
||||
State(api.clone()),
|
||||
HeaderMap::new(),
|
||||
@@ -207,6 +226,7 @@ impl Fixture {
|
||||
} else {
|
||||
Default::default()
|
||||
},
|
||||
..Grants::default()
|
||||
},
|
||||
deadline: tokio::time::Instant::now() + std::time::Duration::from_secs(20),
|
||||
cancellation: CancellationToken::new(),
|
||||
@@ -312,7 +332,7 @@ impl Fixture {
|
||||
}
|
||||
|
||||
fn count(&self, table: &str) -> i64 {
|
||||
assert!(["plugin_invocations", "plugin_effect_receipts"].contains(&table));
|
||||
assert!(["plugin_invocations", "plugin_effect_receipts", "plugin_data_cells", "plugin_data_receipts", "plugin_data_purges"].contains(&table));
|
||||
self.api
|
||||
.config_store
|
||||
.with_conn(|conn| {
|
||||
@@ -391,16 +411,49 @@ fn commit_config(
|
||||
tool: &str,
|
||||
binding: Option<&PluginConnectionBinding>,
|
||||
) {
|
||||
let artifact = &installed.installation.artifacts[0];
|
||||
let plugins = json!({"enabled":{"integration":{
|
||||
commit_config_with_ticket_grants(api, installed, tool, binding, None);
|
||||
}
|
||||
|
||||
fn commit_config_with_ticket_grants(
|
||||
api: &WorkspaceApi,
|
||||
installed: &server_api::PluginOperationResponse,
|
||||
tool: &str,
|
||||
binding: Option<&PluginConnectionBinding>,
|
||||
ticket_grants: Option<&Value>,
|
||||
) {
|
||||
commit_config_for_artifact(
|
||||
api,
|
||||
installed,
|
||||
&installed.installation.artifacts[0],
|
||||
tool,
|
||||
binding,
|
||||
ticket_grants,
|
||||
);
|
||||
}
|
||||
|
||||
fn commit_config_for_artifact(
|
||||
api: &WorkspaceApi,
|
||||
installed: &server_api::PluginOperationResponse,
|
||||
artifact: &server_api::PluginArtifact,
|
||||
tool: &str,
|
||||
binding: Option<&PluginConnectionBinding>,
|
||||
ticket_grants: Option<&Value>,
|
||||
) {
|
||||
let mut plugins = json!({"enabled":{"integration":{
|
||||
"installation_id":installed.installation.installation_id,
|
||||
"expected_last_operation":installed.operation_id,
|
||||
"package_digest":artifact.package_digest,"artifact_digest":artifact.artifact_digest,
|
||||
"tools":[tool],"logging":false,"config":{},
|
||||
"connections":if binding.is_some() { json!({"api":"remote"}) } else { json!({}) }
|
||||
}},"bindings":if let Some(binding) = binding { json!({"remote":binding}) } else { json!({}) }});
|
||||
let selection = json!({"selections":[{"installation_id":installed.installation.installation_id,
|
||||
let mut selection = json!({"selections":[{"installation_id":installed.installation.installation_id,
|
||||
"tools":[tool],"logging":false,"https_slots":if binding.is_some() { vec!["api"] } else { vec![] }}]});
|
||||
if let Some(grants) = ticket_grants {
|
||||
for (key, value) in grants.as_object().unwrap() {
|
||||
plugins["enabled"]["integration"][key] = value.clone();
|
||||
selection["selections"][0][key] = value.clone();
|
||||
}
|
||||
}
|
||||
let content = format!(
|
||||
r#"{{
|
||||
plugins = {};
|
||||
@@ -1250,6 +1303,21 @@ async fn remote_owner_for(
|
||||
f: &Fixture,
|
||||
worker_id: &str,
|
||||
workspace: &str,
|
||||
) -> (Arc<dyn PluginHostOwner>, ServerTask) {
|
||||
remote_owner_for_with_limit(
|
||||
f,
|
||||
worker_id,
|
||||
workspace,
|
||||
yoi_plugin_runtime::owner::PLUGIN_HOST_RESPONSE_BODY_LIMIT,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn remote_owner_for_with_limit(
|
||||
f: &Fixture,
|
||||
worker_id: &str,
|
||||
workspace: &str,
|
||||
response_limit: usize,
|
||||
) -> (Arc<dyn PluginHostOwner>, ServerTask) {
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let address = listener.local_addr().unwrap();
|
||||
@@ -1259,7 +1327,7 @@ async fn remote_owner_for(
|
||||
}));
|
||||
let client = server_api::ServerApiClient::builder(format!("http://{address}"))
|
||||
.unwrap()
|
||||
.response_body_limit(yoi_plugin_runtime::owner::PLUGIN_HOST_RESPONSE_BODY_LIMIT)
|
||||
.response_body_limit(response_limit)
|
||||
.client(
|
||||
reqwest::Client::builder()
|
||||
.no_proxy()
|
||||
|
||||
@@ -311,6 +311,9 @@ pub(super) mod plugin_runtime_tests {
|
||||
.plugins
|
||||
.selections
|
||||
.push(manifest::PluginProfileSelection {
|
||||
ticket_read: false,
|
||||
ticket_data_read: vec![],
|
||||
ticket_data_write: vec![],
|
||||
installation_id: f.installed.installation.installation_id.clone(),
|
||||
tools: vec![f.tool.into()],
|
||||
logging: false,
|
||||
@@ -834,6 +837,9 @@ pub(super) mod plugin_runtime_tests {
|
||||
.plugins
|
||||
.selections
|
||||
.push(manifest::PluginProfileSelection {
|
||||
ticket_read: false,
|
||||
ticket_data_read: vec![],
|
||||
ticket_data_write: vec![],
|
||||
installation_id: f.installed.installation.installation_id.clone(),
|
||||
tools: vec![f.tool.into()],
|
||||
logging: false,
|
||||
|
||||
@@ -129,7 +129,7 @@ pub use ticket::{
|
||||
};
|
||||
|
||||
const OLDEST_SCHEMA_VERSION: i64 = 50;
|
||||
const LATEST_SCHEMA_VERSION: i64 = 110;
|
||||
const LATEST_SCHEMA_VERSION: i64 = 111;
|
||||
const WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME: &str = "durable Workspace Drive grants";
|
||||
const WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME: &str = "Workspace config grants and logical Workdirs";
|
||||
const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline";
|
||||
@@ -487,6 +487,11 @@ const MIGRATIONS: &[Migration] = &[
|
||||
name: "Worker Plugin launch selection associations",
|
||||
apply: migrate_plugin_launches_v109_to_v110,
|
||||
},
|
||||
Migration {
|
||||
version: 111,
|
||||
name: "Typed Plugin Ticket extension data and atomic mutation receipts",
|
||||
apply: migrate_plugin_data_v110_to_v111,
|
||||
},
|
||||
];
|
||||
|
||||
// Both schema-103 identities shipped. Keep the alternate implementation and accept
|
||||
@@ -15897,6 +15902,49 @@ fn migrate_subject_assignments_v107_to_v108(conn: &Connection) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn verify_plugin_data_schema(conn: &Connection) -> Result<()> {
|
||||
let expected = Connection::open_in_memory()?;
|
||||
expected.execute_batch(crate::plugin_receipt::SCHEMA_SQL)?;
|
||||
expected.execute_batch(crate::plugin_data::SCHEMA_SQL)?;
|
||||
let mut query = expected.prepare("SELECT name,sql FROM sqlite_schema WHERE sql IS NOT NULL AND (name LIKE 'plugin_data_%' OR name='plugin_effect_data_collision') ORDER BY name")?;
|
||||
for item in query.query_map([], |row| {
|
||||
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
|
||||
})? {
|
||||
let (name, sql) = item?;
|
||||
let actual: Option<String> = conn
|
||||
.query_row(
|
||||
"SELECT sql FROM sqlite_schema WHERE name=?1",
|
||||
[&name],
|
||||
|row| row.get(0),
|
||||
)
|
||||
.optional()?;
|
||||
if actual.as_deref() != Some(&sql) {
|
||||
return Err(Error::Store(format!(
|
||||
"Plugin extension-data schema has drifted: {name}"
|
||||
)));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn migrate_plugin_data_v110_to_v111(conn: &Connection) -> Result<()> {
|
||||
if current_schema_version(conn)? != 110 {
|
||||
return Err(Error::Store(
|
||||
"Plugin extension-data migration requires schema 110".into(),
|
||||
));
|
||||
}
|
||||
verify_schema_history(conn, 110)?;
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
tx.execute_batch(crate::plugin_data::REFERENCE_MIGRATION_SQL)?;
|
||||
tx.execute_batch(crate::plugin_data::SCHEMA_SQL)?;
|
||||
tx.execute(
|
||||
"INSERT INTO __yoi_schema_migrations(version,name) VALUES(111,?1)",
|
||||
["Typed Plugin Ticket extension data and atomic mutation receipts"],
|
||||
)?;
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn migrate_plugin_launches_v109_to_v110(conn: &Connection) -> Result<()> {
|
||||
if current_schema_version(conn)? != 109 {
|
||||
return Err(Error::Store(
|
||||
@@ -15975,6 +16023,8 @@ fn create_workspace_schema_structure(conn: &Connection) -> Result<()> {
|
||||
conn.execute_batch(crate::plugin_plan::SCHEMA_SQL)?;
|
||||
conn.execute_batch(crate::plugin_receipt::SCHEMA_SQL)?;
|
||||
conn.execute_batch(crate::plugin_plan::LAUNCH_SCHEMA_SQL)?;
|
||||
conn.execute_batch(crate::plugin_data::REFERENCE_MIGRATION_SQL)?;
|
||||
conn.execute_batch(crate::plugin_data::SCHEMA_SQL)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -17015,7 +17065,8 @@ fn apply_migrations_with_secret_source(
|
||||
}
|
||||
|
||||
verify_schema_history(conn, LATEST_SCHEMA_VERSION)?;
|
||||
crate::plugin_catalog::verify(conn)?;
|
||||
crate::plugin_catalog::verify_with_data(conn)?;
|
||||
verify_plugin_data_schema(conn)?;
|
||||
crate::plugin_plan::verify_plan_schema(conn)?;
|
||||
crate::plugin_host::verify_receipt_schema(conn)?;
|
||||
crate::plugin_plan::verify_launch_schema(conn)?;
|
||||
@@ -19304,6 +19355,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
|
||||
include!("ticket_post_upgrade_tests.rs");
|
||||
include!("migration_entrypoint_tests.rs");
|
||||
include!("plugin_host_migration_tests.rs");
|
||||
include!("plugin_data_migration_tests.rs");
|
||||
include!("legacy_evidence_column_tests.rs");
|
||||
|
||||
#[test]
|
||||
@@ -19721,6 +19773,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
|
||||
WorkspaceSchemaMigrationStep {version:108, name:"Subject Ticket responsibility and current-body notification routing".into()},
|
||||
WorkspaceSchemaMigrationStep {version:109, name:"Durable Plugin Host invocations and HTTPS effect receipts".into()},
|
||||
WorkspaceSchemaMigrationStep {version:110, name:"Worker Plugin launch selection associations".into()},
|
||||
WorkspaceSchemaMigrationStep {version:111, name:"Typed Plugin Ticket extension data and atomic mutation receipts".into()},
|
||||
]
|
||||
);
|
||||
|
||||
@@ -19841,6 +19894,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
|
||||
(108, "Subject Ticket responsibility and current-body notification routing".to_string()),
|
||||
(109, "Durable Plugin Host invocations and HTTPS effect receipts".to_string()),
|
||||
(110, "Worker Plugin launch selection associations".to_string()),
|
||||
(111, "Typed Plugin Ticket extension data and atomic mutation receipts".to_string()),
|
||||
]
|
||||
);
|
||||
assert!(!table_exists(conn, "trusted_runtime_records")?);
|
||||
@@ -23132,7 +23186,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
|
||||
(107, "Immutable Worker Plugin plans and retained pins".into()),
|
||||
(108, "Subject Ticket responsibility and current-body notification routing".into()),
|
||||
(109, "Durable Plugin Host invocations and HTTPS effect receipts".into()),
|
||||
(110, "Worker Plugin launch selection associations".into())]);
|
||||
(110, "Worker Plugin launch selection associations".into()),
|
||||
(111, "Typed Plugin Ticket extension data and atomic mutation receipts".into())]);
|
||||
for (table, rows) in unchanged.iter().zip(before.iter()) {
|
||||
assert_eq!(&stored_fixture_rows(conn, table), rows, "preserve {table}");
|
||||
}
|
||||
|
||||
@@ -1071,7 +1071,7 @@ mod subject_assignment_migration_tests {
|
||||
);
|
||||
assert_eq!(history.len(), host.len() + expected.len());
|
||||
assert_component14_append(conn, component);
|
||||
crate::plugin_catalog::verify(conn).unwrap();
|
||||
crate::plugin_catalog::verify_with_data(conn).unwrap();
|
||||
plugin_plan::verify_plan_schema(conn).unwrap();
|
||||
verify(conn).unwrap();
|
||||
assert_health(conn);
|
||||
@@ -1130,6 +1130,10 @@ mod subject_assignment_migration_tests {
|
||||
};
|
||||
tail.push((109, "Durable Plugin Host invocations and HTTPS effect receipts".into()));
|
||||
tail.push((110, "Worker Plugin launch selection associations".into()));
|
||||
tail.push((
|
||||
111,
|
||||
"Typed Plugin Ticket extension data and atomic mutation receipts".into(),
|
||||
));
|
||||
tail
|
||||
}
|
||||
|
||||
@@ -1179,7 +1183,7 @@ mod subject_assignment_migration_tests {
|
||||
fn plugin_schema(conn: &Connection) -> Rows {
|
||||
fixture_query_rows(
|
||||
conn,
|
||||
"SELECT type,name,tbl_name,rootpage,sql FROM sqlite_schema WHERE tbl_name LIKE 'plugin_%' AND tbl_name NOT IN ('plugin_invocations','plugin_effect_receipts') ORDER BY type,name",
|
||||
"SELECT type,name,tbl_name,rootpage,sql FROM sqlite_schema WHERE tbl_name LIKE 'plugin_%' AND tbl_name NOT LIKE 'plugin_data_%' AND tbl_name NOT IN ('plugin_invocations','plugin_effect_receipts','plugin_references') ORDER BY type,name",
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1189,6 +1193,9 @@ mod subject_assignment_migration_tests {
|
||||
let before = snapshot(&conn);
|
||||
let rows = data(&conn, true);
|
||||
let plans = plugin_schema(&conn);
|
||||
// Migration111 intentionally rebuilds the reference guard to admit
|
||||
// data owners; its complete row set must still survive unchanged.
|
||||
let references = stored_fixture_rows(&conn, "plugin_references");
|
||||
let contracts = principal_table_contracts(&conn);
|
||||
let host = host_history(&conn);
|
||||
let component = component_history(&conn);
|
||||
@@ -1205,6 +1212,8 @@ mod subject_assignment_migration_tests {
|
||||
assert_data(&conn, &rows);
|
||||
assert_final(&conn, 107, &host, &component);
|
||||
assert_eq!(plugin_schema(&conn), plans);
|
||||
assert_eq!(stored_fixture_rows(&conn, "plugin_references"), references);
|
||||
verify_plugin_data_schema(&conn).unwrap();
|
||||
assert_eq!(principal_table_contracts(&conn), contracts);
|
||||
let final_snapshot = snapshot(&conn);
|
||||
drop(conn);
|
||||
|
||||
@@ -16,6 +16,11 @@ use crate::{Error, Result};
|
||||
/// Explicit domain-owned purge inventory. The deletion operation tombstone is intentionally
|
||||
/// excluded so retries and audit remain available after the Workspace row is gone.
|
||||
const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
|
||||
"plugin_data_receipts",
|
||||
"plugin_data_purges",
|
||||
"plugin_data_holders",
|
||||
"plugin_data_cells",
|
||||
"plugin_data_cursors",
|
||||
"plugin_effect_receipts",
|
||||
"plugin_invocations",
|
||||
"worker_plugin_launches",
|
||||
@@ -512,6 +517,8 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
|
||||
// commit; update guards remain installed throughout.
|
||||
let mut ticket_delete_guards = Vec::new();
|
||||
for (table, trigger) in [
|
||||
("plugin_data_receipts", "plugin_data_receipt_retained"),
|
||||
("plugin_data_purges", "plugin_data_purge_retained"),
|
||||
("plugin_effect_receipts", "plugin_effect_retained"),
|
||||
("plugin_invocations", "plugin_invocation_retained"),
|
||||
("worker_plugin_launches", "worker_plugin_launch_retained"),
|
||||
|
||||
Reference in New Issue
Block a user