Skip to main content

syncular_client/
previous_version.rs

1//! RFC 0005 previous-version context — phase 1 (the storage half).
2//!
3//! A schema bump wipes the local replica (§7.4.3), so the rows the app could
4//! see a moment ago are gone before the server can re-bootstrap them. This
5//! module persists the SEMANTIC schema descriptor (D1), captures a bounded,
6//! typed, read-only copy of the pre-reset rows into a SECOND database file
7//! beside the replica (D2/D3/A4), and discards that file.
8//!
9//! Three rules shape the implementation:
10//!
11//! - Storage is a sibling FILE `<replica path>.prev-context`, never a table in
12//!   the replica connection. The guarantee is stated narrowly: an older
13//!   client's ordinary query connection does not attach this file, so no SQL it
14//!   runs can reach the container. It is NOT protection against arbitrary
15//!   filesystem access, and the filename is a code constant (not a secret).
16//! - The capture is typed by the OLD schema's [`LocalSchemaDescriptor`]. SQLite
17//!   affinity cannot recover a semantic type (`boolean` and `integer` are both
18//!   INTEGER; `string`, `json` and `blob_ref` are all TEXT), so a
19//!   descriptor-free capture would be undecodable. It is NEVER inferred.
20//! - Every budget is measured BEFORE any row is materialized. An over-budget
21//!   capture stores nothing and leaves no file behind.
22//!
23//! Phase 2 (commands, bridges, conformance) layers the read surface, the
24//! compatibility audit and the lifetime triggers on top; this module owns only
25//! the descriptor, the container file and its capture/discard semantics.
26
27use std::collections::BTreeSet;
28
29use rusqlite::types::ValueRef;
30use rusqlite::{Connection, OptionalExtension};
31use serde::Serialize;
32use serde_json::{json, Map, Value};
33use ssp2::segment::{Column, ColumnType};
34
35use crate::client::{meta_delete, meta_get, meta_set, quote_ident, sql_ref_to_json};
36use crate::schema::ClientSchema;
37use crate::values::{bytes_to_hex, column_value_to_json, json_to_column_value};
38
39/// `_syncular_meta` key holding the persisted [`LocalSchemaDescriptor`] (D1).
40pub const LOCAL_SCHEMA_DESCRIPTOR_KEY: &str = "localSchemaDescriptor";
41/// `_syncular_meta` key holding only a capture REFUSAL. The successful record
42/// lives inside the container file, so this key is absent when a container is
43/// present. It stays in the replica because it is small and typed.
44pub const PREVIOUS_VERSION_CONTEXT_KEY: &str = "previousVersionContext";
45/// `_syncular_meta` key holding the pre-reset compatibility audit (D6, phase 2).
46/// Declared here because a discard clears it.
47pub const PREVIOUS_VERSION_AUDIT_KEY: &str = "previousVersionAudit";
48
49/// RFC 0005 D3: the container is the sibling file `<replica path>.prev-context`.
50/// The name is code-derived and never persisted in the replica database.
51pub const PREVIOUS_VERSION_CONTAINER_SUFFIX: &str = ".prev-context";
52
53/// The one non-reserved row table INSIDE the container file (D3).
54const CONTAINER_TABLE: &str = "syncular_prev_context";
55/// Container-local metadata table holding the [`PreviousVersionRecord`].
56const CONTAINER_META_TABLE: &str = "_syncular_prev_context_meta";
57
58const DESCRIPTOR_VERSION: i64 = 1;
59const RECORD_VERSION: i64 = 1;
60/// Rows copied per `INSERT` batch (A4 step 5).
61const COPY_BATCH_ROWS: i64 = 500;
62
63/// D2/D8/A4: capture bounds. Every bound is measured before materializing.
64#[derive(Debug, Clone, Copy, PartialEq, Eq)]
65pub struct PreviousVersionContextConfig {
66    /// Feature flag, default off. A feature flag, never a security control.
67    pub enabled: bool,
68    pub max_bytes: i64,
69    pub max_rows: i64,
70    pub max_tables: i64,
71    pub max_row_bytes: i64,
72    /// D7/A1 TTL. Enforced ONLY by an aware binary, at boot and at read. It is
73    /// a hygiene bound for aware binaries and says NOTHING about an unaware
74    /// same-schema rollback, which runs none of this code and therefore leaves
75    /// the container in place indefinitely.
76    pub max_age_ms: i64,
77}
78
79impl Default for PreviousVersionContextConfig {
80    fn default() -> Self {
81        Self {
82            enabled: false,
83            max_bytes: 8 * 1024 * 1024,
84            max_rows: 20_000,
85            max_tables: 32,
86            max_row_bytes: 1024 * 1024,
87            max_age_ms: 24 * 60 * 60 * 1000,
88        }
89    }
90}
91
92impl PreviousVersionContextConfig {
93    /// D8: every bound is a positive safe integer — the same validation the TS
94    /// core runs on `previousVersionContext`, with the same error code.
95    pub fn validate(&self) -> Result<(), String> {
96        let bounds = [
97            ("maxBytes", self.max_bytes),
98            ("maxRows", self.max_rows),
99            ("maxTables", self.max_tables),
100            ("maxRowBytes", self.max_row_bytes),
101            ("maxAgeMs", self.max_age_ms),
102        ];
103        for (name, value) in bounds {
104            if value <= 0 || u64::try_from(value).map_or(true, |v| v > MAX_SAFE_INTEGER) {
105                return Err(format!(
106                    "sync.invalid_request: previousVersionContext.{name} must be a positive safe integer"
107                ));
108            }
109        }
110        Ok(())
111    }
112}
113
114/// JavaScript's `Number.MAX_SAFE_INTEGER`; a Rust host crosses a JSON boundary,
115/// so a bound past it could not survive the round trip.
116const MAX_SAFE_INTEGER: u64 = 9_007_199_254_740_991;
117
118fn local_corrupt(what: &str) -> String {
119    format!("sync.local_corrupt: persisted {what} is invalid")
120}
121
122/// The sibling container path for a file-backed replica. In-memory replicas
123/// have no path and therefore no container (the feature resolves as off).
124pub fn previous_version_container_path(replica_path: &str) -> String {
125    format!("{replica_path}{PREVIOUS_VERSION_CONTAINER_SUFFIX}")
126}
127
128// ---------------------------------------------------------------------------
129// Descriptor (D1)
130// ---------------------------------------------------------------------------
131
132#[derive(Debug, Clone, PartialEq, Eq)]
133pub struct DescriptorColumn {
134    pub name: String,
135    /// The SEMANTIC local type (`localColumnType`): for an encrypted column
136    /// that is its declared app type, because the local mirror is plaintext.
137    pub ty: ColumnType,
138}
139
140#[derive(Debug, Clone, PartialEq, Eq)]
141pub struct DescriptorTable {
142    pub name: String,
143    pub primary_key: String,
144    pub columns: Vec<DescriptorColumn>,
145}
146
147#[derive(Debug, Clone, PartialEq, Eq)]
148pub struct LocalSchemaDescriptor {
149    pub version: i32,
150    pub tables: Vec<DescriptorTable>,
151}
152
153impl LocalSchemaDescriptor {
154    pub fn table(&self, name: &str) -> Option<&DescriptorTable> {
155        self.tables.iter().find(|table| table.name == name)
156    }
157}
158
159/// D1: the semantic local types of every table in the running generated
160/// schema, built from the compiled LOCAL columns — never from SQLite affinity.
161pub fn build_local_schema_descriptor(schema: &ClientSchema) -> LocalSchemaDescriptor {
162    LocalSchemaDescriptor {
163        version: schema.version,
164        tables: schema
165            .tables
166            .iter()
167            .map(|table| DescriptorTable {
168                name: table.name.clone(),
169                primary_key: table.primary_key.clone(),
170                columns: table
171                    .columns
172                    .iter()
173                    .map(|column| DescriptorColumn {
174                        name: column.name.clone(),
175                        ty: column.ty,
176                    })
177                    .collect(),
178            })
179            .collect(),
180    }
181}
182
183fn encode_local_schema_descriptor(descriptor: &LocalSchemaDescriptor) -> String {
184    let tables = descriptor
185        .tables
186        .iter()
187        .map(|table| {
188            let columns = table
189                .columns
190                .iter()
191                .map(|column| json!({ "name": column.name, "type": column.ty.name() }))
192                .collect::<Vec<_>>();
193            json!({
194                "name": table.name,
195                "primaryKey": table.primary_key,
196                "columns": columns,
197            })
198        })
199        .collect::<Vec<_>>();
200    json!({
201        "v": DESCRIPTOR_VERSION,
202        "version": descriptor.version,
203        "tables": tables,
204    })
205    .to_string()
206}
207
208/// Strict shape comparison: exactly these keys, nothing else. Equivalent to
209/// the TS core's sorted-key/length check.
210fn has_exact_keys(object: &Map<String, Value>, expected: &[&str]) -> bool {
211    object.len() == expected.len() && expected.iter().all(|key| object.contains_key(*key))
212}
213
214/// A non-negative safe integer, mirroring the TS `isCount` predicate.
215fn as_count(value: Option<&Value>) -> Option<i64> {
216    value
217        .and_then(Value::as_i64)
218        .filter(|value| *value >= 0 && u64::try_from(*value).is_ok_and(|v| v <= MAX_SAFE_INTEGER))
219}
220
221fn decode_descriptor_column(value: &Value) -> Result<DescriptorColumn, String> {
222    let Value::Object(column) = value else {
223        return Err(local_corrupt("schema descriptor"));
224    };
225    if !has_exact_keys(column, &["name", "type"]) {
226        return Err(local_corrupt("schema descriptor"));
227    }
228    let (Some(Value::String(name)), Some(Value::String(type_name))) =
229        (column.get("name"), column.get("type"))
230    else {
231        return Err(local_corrupt("schema descriptor"));
232    };
233    let ty = ColumnType::from_name(type_name).ok_or_else(|| local_corrupt("schema descriptor"))?;
234    Ok(DescriptorColumn {
235        name: name.clone(),
236        ty,
237    })
238}
239
240fn decode_descriptor_table(value: &Value) -> Result<DescriptorTable, String> {
241    let Value::Object(table) = value else {
242        return Err(local_corrupt("schema descriptor"));
243    };
244    if !has_exact_keys(table, &["columns", "name", "primaryKey"]) {
245        return Err(local_corrupt("schema descriptor"));
246    }
247    let (Some(Value::String(name)), Some(Value::String(primary_key)), Some(Value::Array(columns))) = (
248        table.get("name"),
249        table.get("primaryKey"),
250        table.get("columns"),
251    ) else {
252        return Err(local_corrupt("schema descriptor"));
253    };
254    Ok(DescriptorTable {
255        name: name.clone(),
256        primary_key: primary_key.clone(),
257        columns: columns
258            .iter()
259            .map(decode_descriptor_column)
260            .collect::<Result<Vec<_>, _>>()?,
261    })
262}
263
264/// Strict decode: an unknown shape is corruption, never a best guess.
265pub fn decode_local_schema_descriptor(raw: &str) -> Result<LocalSchemaDescriptor, String> {
266    let parsed: Value =
267        serde_json::from_str(raw).map_err(|_| local_corrupt("schema descriptor"))?;
268    let Value::Object(record) = &parsed else {
269        return Err(local_corrupt("schema descriptor"));
270    };
271    if !has_exact_keys(record, &["tables", "v", "version"])
272        || as_count(record.get("v")) != Some(DESCRIPTOR_VERSION)
273    {
274        return Err(local_corrupt("schema descriptor"));
275    }
276    let version = as_count(record.get("version"))
277        .and_then(|value| i32::try_from(value).ok())
278        .ok_or_else(|| local_corrupt("schema descriptor"))?;
279    let Some(Value::Array(tables)) = record.get("tables") else {
280        return Err(local_corrupt("schema descriptor"));
281    };
282    Ok(LocalSchemaDescriptor {
283        version,
284        tables: tables
285            .iter()
286            .map(decode_descriptor_table)
287            .collect::<Result<Vec<_>, _>>()?,
288    })
289}
290
291/// D1: persist the descriptor. Callers write this beside every write of
292/// [`LOCAL_SCHEMA_VERSION_KEY`]; a descriptor without a matching marker is
293/// treated as absent at reset time.
294pub fn set_local_schema_descriptor(
295    conn: &Connection,
296    schema: &ClientSchema,
297) -> rusqlite::Result<()> {
298    conn.execute(
299        "INSERT OR REPLACE INTO _syncular_meta(key,value) VALUES (?1,?2)",
300        rusqlite::params![
301            LOCAL_SCHEMA_DESCRIPTOR_KEY,
302            encode_local_schema_descriptor(&build_local_schema_descriptor(schema))
303        ],
304    )?;
305    Ok(())
306}
307
308/// Load and strictly decode the stored descriptor; `None` when absent/corrupt.
309pub fn load_local_schema_descriptor(conn: &Connection) -> Option<LocalSchemaDescriptor> {
310    let raw = meta_get(conn, LOCAL_SCHEMA_DESCRIPTOR_KEY)?;
311    decode_local_schema_descriptor(&raw).ok()
312}
313
314// ---------------------------------------------------------------------------
315// Container file (D3)
316// ---------------------------------------------------------------------------
317
318/// Drop both container tables. Also the unconditional orphan sweep (D5).
319pub fn drop_previous_version_container(conn: &Connection) -> Result<(), String> {
320    conn.execute_batch(&format!(
321        "DROP TABLE IF EXISTS {CONTAINER_TABLE}; DROP TABLE IF EXISTS {CONTAINER_META_TABLE};"
322    ))
323    .map_err(|error| error.to_string())
324}
325
326/// Successful capture record, stored INSIDE the container file.
327#[derive(Debug, Clone, PartialEq, Eq)]
328pub struct PreviousVersionRecord {
329    pub previous_version: i32,
330    pub current_version: i32,
331    pub tables: Vec<DescriptorTable>,
332    pub rows: i64,
333    pub bytes: i64,
334    pub created_at_ms: i64,
335}
336
337impl PreviousVersionRecord {
338    /// The captured shape of one table — what a read needs to decode payloads.
339    pub fn table(&self, name: &str) -> Option<&DescriptorTable> {
340        self.tables.iter().find(|table| table.name == name)
341    }
342}
343
344fn decode_record(raw: &str) -> Result<PreviousVersionRecord, String> {
345    let parsed: Value =
346        serde_json::from_str(raw).map_err(|_| local_corrupt("previous-version context"))?;
347    let Value::Object(record) = &parsed else {
348        return Err(local_corrupt("previous-version context"));
349    };
350    if !has_exact_keys(
351        record,
352        &[
353            "bytes",
354            "createdAtMs",
355            "currentVersion",
356            "previousVersion",
357            "rows",
358            "tables",
359            "v",
360        ],
361    ) || as_count(record.get("v")) != Some(RECORD_VERSION)
362    {
363        return Err(local_corrupt("previous-version context"));
364    }
365    let count = |key: &str| {
366        as_count(record.get(key))
367            .and_then(|value| i32::try_from(value).ok())
368            .ok_or_else(|| local_corrupt("previous-version context"))
369    };
370    let Some(Value::Array(tables)) = record.get("tables") else {
371        return Err(local_corrupt("previous-version context"));
372    };
373    Ok(PreviousVersionRecord {
374        previous_version: count("previousVersion")?,
375        current_version: count("currentVersion")?,
376        tables: tables
377            .iter()
378            .map(decode_descriptor_table)
379            .collect::<Result<Vec<_>, _>>()?,
380        rows: as_count(record.get("rows"))
381            .ok_or_else(|| local_corrupt("previous-version context"))?,
382        bytes: as_count(record.get("bytes"))
383            .ok_or_else(|| local_corrupt("previous-version context"))?,
384        created_at_ms: as_count(record.get("createdAtMs"))
385            .ok_or_else(|| local_corrupt("previous-version context"))?,
386    })
387}
388
389fn encode_record(record: &PreviousVersionRecord) -> String {
390    let tables = record
391        .tables
392        .iter()
393        .map(|table| {
394            let columns = table
395                .columns
396                .iter()
397                .map(|column| json!({ "name": column.name, "type": column.ty.name() }))
398                .collect::<Vec<_>>();
399            json!({
400                "name": table.name,
401                "primaryKey": table.primary_key,
402                "columns": columns,
403            })
404        })
405        .collect::<Vec<_>>();
406    json!({
407        "v": RECORD_VERSION,
408        "previousVersion": record.previous_version,
409        "currentVersion": record.current_version,
410        "tables": tables,
411        "rows": record.rows,
412        "bytes": record.bytes,
413        "createdAtMs": record.created_at_ms,
414    })
415    .to_string()
416}
417
418fn write_record(conn: &Connection, record: &PreviousVersionRecord) -> Result<(), String> {
419    conn.execute_batch(&format!(
420        "CREATE TABLE IF NOT EXISTS {CONTAINER_META_TABLE} (
421           id INTEGER PRIMARY KEY CHECK (id = 1),
422           record TEXT NOT NULL);
423         DELETE FROM {CONTAINER_META_TABLE};"
424    ))
425    .map_err(|error| error.to_string())?;
426    conn.execute(
427        &format!("INSERT INTO {CONTAINER_META_TABLE}(id, record) VALUES (1, ?1)"),
428        rusqlite::params![encode_record(record)],
429    )
430    .map_err(|error| error.to_string())?;
431    Ok(())
432}
433
434/// Read the container's own metadata record; `None` when absent/corrupt.
435pub fn read_previous_version_container(
436    conn: &Connection,
437) -> Result<Option<PreviousVersionRecord>, String> {
438    let present: i64 = conn
439        .query_row(
440            "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?1",
441            rusqlite::params![CONTAINER_META_TABLE],
442            |row| row.get(0),
443        )
444        .map_err(|error| error.to_string())?;
445    if present == 0 {
446        return Ok(None);
447    }
448    let raw: Option<String> = conn
449        .query_row(
450            &format!("SELECT record FROM {CONTAINER_META_TABLE} WHERE id = 1"),
451            [],
452            |row| row.get(0),
453        )
454        .optional()
455        .map_err(|error| error.to_string())?;
456    Ok(raw.and_then(|raw| decode_record(&raw).ok()))
457}
458
459/// D7: read one previous table from the container, `limit + 1` rows so the
460/// caller can report truncation. `rowIds` empty means every row of the table.
461pub fn read_previous_version_rows(
462    conn: &Connection,
463    table: &DescriptorTable,
464    row_ids: &[String],
465    limit: i64,
466) -> Result<(Vec<Map<String, Value>>, bool), String> {
467    let filter = if row_ids.is_empty() {
468        String::new()
469    } else {
470        let placeholders = (0..row_ids.len())
471            .map(|index| format!("?{}", index + 2))
472            .collect::<Vec<_>>()
473            .join(", ");
474        format!(" AND row_id IN ({placeholders})")
475    };
476    let mut params: Vec<rusqlite::types::Value> =
477        vec![rusqlite::types::Value::Text(table.name.clone())];
478    params.extend(
479        row_ids
480            .iter()
481            .map(|row_id| rusqlite::types::Value::Text(row_id.clone())),
482    );
483    params.push(rusqlite::types::Value::Integer(limit + 1));
484    let limit_placeholder = params.len();
485    let mut stmt = conn
486        .prepare(&format!(
487            "SELECT payload FROM {CONTAINER_TABLE}
488               WHERE tbl = ?1{filter} ORDER BY row_id ASC LIMIT ?{limit_placeholder}"
489        ))
490        .map_err(|error| error.to_string())?;
491    let payloads = stmt
492        .query_map(rusqlite::params_from_iter(params), |row| {
493            row.get::<_, String>(0)
494        })
495        .map_err(|error| error.to_string())?
496        .collect::<Result<Vec<_>, _>>()
497        .map_err(|error| error.to_string())?;
498    let truncated = payloads.len() as i64 > limit;
499    let visible = if truncated {
500        limit as usize
501    } else {
502        payloads.len()
503    };
504    let mut rows = Vec::with_capacity(visible);
505    for payload in payloads.into_iter().take(visible) {
506        rows.push(payload_to_row(table, &payload)?);
507    }
508    Ok((rows, truncated))
509}
510
511// ---------------------------------------------------------------------------
512// Bounded capture (D2 / A4)
513// ---------------------------------------------------------------------------
514
515/// D2/A4: measure the capture BEFORE materializing anything. Ordered probes,
516/// aborting on the first violation: table count; per-table row count with a
517/// `LIMIT maxRows + 1` early-abort probe; a single-row probe for rows larger
518/// than `maxRowBytes`; a bounded `SUM` over a `LIMIT`ed subquery. Only then is
519/// anything copied.
520#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
521pub struct CaptureMeasurement {
522    pub tables: i64,
523    pub rows: i64,
524    pub bytes: i64,
525}
526
527#[derive(Debug, Clone, Copy, PartialEq, Eq)]
528pub enum PreviousVersionRefusalReason {
529    NoPreviousDescriptor,
530    CaptureExceededBudget,
531}
532
533impl PreviousVersionRefusalReason {
534    pub fn name(self) -> &'static str {
535        match self {
536            Self::NoPreviousDescriptor => "no-previous-descriptor",
537            Self::CaptureExceededBudget => "capture-exceeded-budget",
538        }
539    }
540
541    fn as_read_reason(self) -> PreviousVersionReason {
542        match self {
543            Self::NoPreviousDescriptor => PreviousVersionReason::NoPreviousDescriptor,
544            Self::CaptureExceededBudget => PreviousVersionReason::CaptureExceededBudget,
545        }
546    }
547}
548
549#[derive(Debug, Clone, Copy, PartialEq, Eq)]
550pub enum CaptureOutcome {
551    Captured(CaptureMeasurement),
552    Refused {
553        reason: PreviousVersionRefusalReason,
554        measurement: CaptureMeasurement,
555    },
556}
557
558fn sum_bytes_expression(columns: &[DescriptorColumn]) -> String {
559    columns
560        .iter()
561        .map(|column| {
562            format!(
563                "COALESCE(LENGTH(CAST({} AS BLOB)), 0)",
564                quote_ident(&column.name)
565            )
566        })
567        .collect::<Vec<_>>()
568        .join(" + ")
569}
570
571fn render_row_id(value: ValueRef<'_>) -> String {
572    match value {
573        ValueRef::Null => String::new(),
574        ValueRef::Integer(value) => value.to_string(),
575        ValueRef::Real(value) => value.to_string(),
576        ValueRef::Text(value) => String::from_utf8_lossy(value).into_owned(),
577        ValueRef::Blob(value) => bytes_to_hex(value),
578    }
579}
580
581/// A descriptor column as the row-value codec's column shape (`nullable` is
582/// irrelevant to the local read, which is where the type is recovered).
583fn column_ref(column: &DescriptorColumn) -> Column {
584    Column {
585        name: column.name.clone(),
586        ty: column.ty,
587        nullable: true,
588    }
589}
590
591fn payload_to_row(table: &DescriptorTable, payload: &str) -> Result<Map<String, Value>, String> {
592    let parsed: Value =
593        serde_json::from_str(payload).map_err(|_| local_corrupt("previous-version payload"))?;
594    let Value::Object(record) = parsed else {
595        return Err(local_corrupt("previous-version payload"));
596    };
597    let mut row = Map::new();
598    for column in &table.columns {
599        let value = json_to_column_value(&column_ref(column), record.get(&column.name))
600            .map_err(|_| local_corrupt("previous-version payload"))?;
601        row.insert(column.name.clone(), column_value_to_json(&value));
602    }
603    Ok(row)
604}
605
606/// D3/D5 steps 3-5: copy the descriptor's tables into the container file, in
607/// one container transaction and [`COPY_BATCH_ROWS`] batches.
608pub fn capture_previous_version(
609    replica: &Connection,
610    container: &mut Connection,
611    descriptor: &LocalSchemaDescriptor,
612    current_version: i32,
613    config: &PreviousVersionContextConfig,
614    now_ms: i64,
615) -> Result<CaptureOutcome, String> {
616    let discovered: BTreeSet<String> = {
617        let mut stmt = replica
618            .prepare(
619                "SELECT name FROM sqlite_master WHERE type = 'table'
620                   AND name NOT LIKE '_syncular_%' AND name NOT LIKE 'sqlite_%'",
621            )
622            .map_err(|error| error.to_string())?;
623        let names = stmt
624            .query_map([], |row| row.get::<_, String>(0))
625            .map_err(|error| error.to_string())?
626            .collect::<Result<BTreeSet<_>, _>>()
627            .map_err(|error| error.to_string())?;
628        names
629    };
630    // Only descriptor-known tables carry semantic types; those are the tables
631    // the copy would hold, and the count the table budget bounds.
632    let tables: Vec<&DescriptorTable> = descriptor
633        .tables
634        .iter()
635        .filter(|table| discovered.contains(&table.name))
636        .collect();
637    let refused = |rows: i64, bytes: i64| CaptureOutcome::Refused {
638        reason: PreviousVersionRefusalReason::CaptureExceededBudget,
639        measurement: CaptureMeasurement {
640            tables: tables.len() as i64,
641            rows,
642            bytes,
643        },
644    };
645    if tables.len() as i64 > config.max_tables {
646        return Ok(refused(0, 0));
647    }
648
649    // Probe 2: rows per table, stopping at maxRows + 1 rather than counting a
650    // hundred-million-row table in full.
651    let mut measured_rows = 0i64;
652    let mut per_table_rows: Vec<i64> = Vec::with_capacity(tables.len());
653    for table in &tables {
654        let count: i64 = replica
655            .query_row(
656                &format!(
657                    "SELECT COUNT(*) FROM (SELECT 1 FROM {} LIMIT ?1)",
658                    quote_ident(&table.name)
659                ),
660                rusqlite::params![config.max_rows + 1],
661                |row| row.get(0),
662            )
663            .map_err(|error| error.to_string())?;
664        measured_rows += count;
665        per_table_rows.push(count);
666        if measured_rows > config.max_rows {
667            return Ok(refused(measured_rows, 0));
668        }
669    }
670
671    // Probes 3-4: one oversized row is rejected before it is ever loaded, then
672    // the byte total is summed over a bounded scan only.
673    let mut measured_bytes = 0i64;
674    for (index, table) in tables.iter().enumerate() {
675        if table.columns.is_empty() {
676            continue;
677        }
678        let byte_sum = sum_bytes_expression(&table.columns);
679        let oversized: Option<i64> = replica
680            .query_row(
681                &format!(
682                    "SELECT 1 FROM {} WHERE ({byte_sum}) > ?1 LIMIT 1",
683                    quote_ident(&table.name)
684                ),
685                rusqlite::params![config.max_row_bytes],
686                |row| row.get(0),
687            )
688            .optional()
689            .map_err(|error| error.to_string())?;
690        if oversized.is_some() {
691            return Ok(refused(measured_rows, measured_bytes));
692        }
693        let total: i64 = replica
694            .query_row(
695                &format!(
696                    "SELECT COALESCE(SUM(bytes), 0) FROM (
697                       SELECT ({byte_sum}) AS bytes FROM {} LIMIT ?1
698                     )",
699                    quote_ident(&table.name)
700                ),
701                rusqlite::params![per_table_rows[index]],
702                |row| row.get(0),
703            )
704            .map_err(|error| error.to_string())?;
705        measured_bytes += total;
706        if measured_bytes > config.max_bytes {
707            return Ok(refused(measured_rows, measured_bytes));
708        }
709    }
710
711    // Step 5: copy in batches, so no step holds a whole table in memory.
712    let mut copied_rows = 0i64;
713    let transaction = container.transaction().map_err(|error| error.to_string())?;
714    drop_previous_version_container(&transaction)?;
715    transaction
716        .execute_batch(&format!(
717            "CREATE TABLE {CONTAINER_TABLE} (
718               tbl TEXT NOT NULL,
719               row_id TEXT NOT NULL,
720               payload TEXT NOT NULL,
721               PRIMARY KEY (tbl, row_id))"
722        ))
723        .map_err(|error| error.to_string())?;
724    for table in &tables {
725        let select_columns = table
726            .columns
727            .iter()
728            .map(|column| quote_ident(&column.name))
729            .collect::<Vec<_>>()
730            .join(", ");
731        let primary_key_index = table
732            .columns
733            .iter()
734            .position(|column| column.name == table.primary_key);
735        let select_sql = format!(
736            "SELECT rowid, {select_columns} FROM {} WHERE rowid > ?1 ORDER BY rowid ASC LIMIT ?2",
737            quote_ident(&table.name)
738        );
739        let mut last_row_id = -1i64;
740        loop {
741            let batch: Vec<(i64, String, String)> = {
742                let mut stmt = replica
743                    .prepare(&select_sql)
744                    .map_err(|error| error.to_string())?;
745                let mut rows = stmt
746                    .query(rusqlite::params![last_row_id, COPY_BATCH_ROWS])
747                    .map_err(|error| error.to_string())?;
748                let mut batch = Vec::new();
749                while let Some(row) = rows.next().map_err(|error| error.to_string())? {
750                    let row_id = primary_key_index.map_or_else(String::new, |index| {
751                        row.get_ref(index + 1)
752                            .map(render_row_id)
753                            .unwrap_or_default()
754                    });
755                    let mut payload = Map::new();
756                    for (index, column) in table.columns.iter().enumerate() {
757                        let raw = row.get_ref(index + 1).map_err(|error| error.to_string())?;
758                        payload.insert(
759                            column.name.clone(),
760                            sql_ref_to_json(&column_ref(column), raw),
761                        );
762                    }
763                    batch.push((
764                        row.get::<_, i64>(0).map_err(|error| error.to_string())?,
765                        row_id,
766                        Value::Object(payload).to_string(),
767                    ));
768                }
769                batch
770            };
771            if batch.is_empty() {
772                break;
773            }
774            {
775                let mut insert = transaction
776                    .prepare(&format!(
777                        "INSERT INTO {CONTAINER_TABLE}(tbl, row_id, payload) VALUES (?1, ?2, ?3)"
778                    ))
779                    .map_err(|error| error.to_string())?;
780                for (_, row_id, payload) in &batch {
781                    insert
782                        .execute(rusqlite::params![table.name, row_id, payload])
783                        .map_err(|error| error.to_string())?;
784                }
785            }
786            copied_rows += batch.len() as i64;
787            last_row_id = batch.last().map_or(last_row_id, |entry| entry.0);
788        }
789    }
790    write_record(
791        &transaction,
792        &PreviousVersionRecord {
793            previous_version: descriptor.version,
794            current_version,
795            tables: tables.iter().map(|table| (*table).clone()).collect(),
796            rows: copied_rows,
797            bytes: measured_bytes,
798            created_at_ms: now_ms,
799        },
800    )?;
801    transaction.commit().map_err(|error| error.to_string())?;
802    Ok(CaptureOutcome::Captured(CaptureMeasurement {
803        tables: tables.len() as i64,
804        rows: copied_rows,
805        bytes: measured_bytes,
806    }))
807}
808
809// ---------------------------------------------------------------------------
810// Durable refusal (the read surface for a capture that stored nothing)
811// ---------------------------------------------------------------------------
812
813/// Record a refusal in `_syncular_meta`, with counts only — never row content.
814fn write_previous_version_refusal(
815    conn: &Connection,
816    reason: PreviousVersionRefusalReason,
817    measurement: CaptureMeasurement,
818) {
819    meta_set(
820        conn,
821        PREVIOUS_VERSION_CONTEXT_KEY,
822        &json!({
823            "v": 1,
824            "reason": reason.name(),
825            "tables": measurement.tables,
826            "rows": measurement.rows,
827            "bytes": measurement.bytes,
828        })
829        .to_string(),
830    );
831}
832
833// ---------------------------------------------------------------------------
834// File-level lifecycle: sweep, capture orchestration, discard
835// ---------------------------------------------------------------------------
836
837/// Best-effort physical removal of the container file. The connection MUST be
838/// closed first; the `-wal`/`-shm` siblings go with it.
839fn remove_previous_version_files(path: &str) {
840    for suffix in ["", "-wal", "-shm"] {
841        let _ = std::fs::remove_file(format!("{path}{suffix}"));
842    }
843}
844
845/// D5 step 1: the unconditional orphan sweep. Drops the container tables,
846/// removes the file, and clears a durable refusal recorded for it, so a
847/// crash-interrupted reset cannot leave a container beside a matching marker.
848/// A no-op when no container file exists.
849///
850/// Everything here is best-effort: the file is the thing being deleted, so a
851/// torn or unreadable container must not refuse its own removal and must not
852/// block the client's open.
853pub fn sweep_previous_version_container(
854    replica: &Connection,
855    replica_path: &str,
856) -> Result<(), String> {
857    let path = previous_version_container_path(replica_path);
858    if !std::path::Path::new(&path).exists() {
859        return Ok(());
860    }
861    if let Ok(container) = Connection::open(&path) {
862        let _ = drop_previous_version_container(&container);
863        let _ = container.close();
864    }
865    remove_previous_version_files(&path);
866    meta_delete(replica, PREVIOUS_VERSION_CONTEXT_KEY);
867    Ok(())
868}
869
870/// D5 steps 2-5: load the stored descriptor, then capture into the container
871/// file. A refused or failed capture stores nothing, records its reason, and
872/// leaves NO container file behind.
873pub fn capture_previous_version_from_replica(
874    replica: &Connection,
875    replica_path: &str,
876    previous_version: i32,
877    current_version: i32,
878    config: &PreviousVersionContextConfig,
879    now_ms: i64,
880) -> Result<CaptureOutcome, String> {
881    let path = previous_version_container_path(replica_path);
882    let descriptor = load_local_schema_descriptor(replica)
883        .filter(|descriptor| descriptor.version == previous_version);
884    let Some(descriptor) = descriptor else {
885        // D1: no fallback and no inference — a database last opened by an
886        // unaware binary has no descriptor, and the capture is refused.
887        let measurement = CaptureMeasurement::default();
888        write_previous_version_refusal(
889            replica,
890            PreviousVersionRefusalReason::NoPreviousDescriptor,
891            measurement,
892        );
893        return Ok(CaptureOutcome::Refused {
894            reason: PreviousVersionRefusalReason::NoPreviousDescriptor,
895            measurement,
896        });
897    };
898    let mut container = Connection::open(&path)
899        .map_err(|error| format!("open previous-version container {path:?}: {error}"))?;
900    let outcome = capture_previous_version(
901        replica,
902        &mut container,
903        &descriptor,
904        current_version,
905        config,
906        now_ms,
907    );
908    let _ = container.close();
909    match outcome {
910        Ok(CaptureOutcome::Captured(measurement)) => {
911            meta_delete(replica, PREVIOUS_VERSION_CONTEXT_KEY);
912            Ok(CaptureOutcome::Captured(measurement))
913        }
914        Ok(CaptureOutcome::Refused {
915            reason,
916            measurement,
917        }) => {
918            write_previous_version_refusal(replica, reason, measurement);
919            remove_previous_version_files(&path);
920            Ok(CaptureOutcome::Refused {
921                reason,
922                measurement,
923            })
924        }
925        Err(error) => {
926            remove_previous_version_files(&path);
927            Err(error)
928        }
929    }
930}
931
932/// D7: `limit` defaults to 50 and is capped at 200.
933pub const PREVIOUS_VERSION_DEFAULT_LIMIT: i64 = 50;
934pub const PREVIOUS_VERSION_MAX_LIMIT: i64 = 200;
935
936/// D7: the read surface. `state` is always `previousVersion`, never
937/// `complete`; the container never contributes to `querySnapshot().coverage`.
938#[derive(Debug, Clone, PartialEq, Serialize)]
939#[serde(rename_all = "camelCase")]
940pub struct PreviousVersionSnapshot {
941    pub state: &'static str,
942    pub available: bool,
943    #[serde(skip_serializing_if = "Option::is_none")]
944    pub previous_version: Option<i32>,
945    pub current_version: i32,
946    #[serde(skip_serializing_if = "Option::is_none")]
947    pub reason: Option<PreviousVersionReason>,
948    pub rows: Vec<Map<String, Value>>,
949    pub truncated: bool,
950}
951
952impl PreviousVersionSnapshot {
953    fn unavailable(current_version: i32, reason: PreviousVersionReason) -> Self {
954        Self {
955            state: PREVIOUS_VERSION_STATE,
956            available: false,
957            previous_version: None,
958            current_version,
959            reason: Some(reason),
960            rows: Vec::new(),
961            truncated: false,
962        }
963    }
964}
965
966/// The one value `state` ever takes (D7).
967pub const PREVIOUS_VERSION_STATE: &str = "previousVersion";
968
969/// Every reason the read surface can name.
970#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
971#[serde(rename_all = "kebab-case")]
972pub enum PreviousVersionReason {
973    NotConfigured,
974    NoPreviousDescriptor,
975    CaptureExceededBudget,
976    CoverageComplete,
977    Expired,
978    LeaseInactive,
979    ScopeRevoked,
980}
981
982/// D7: the in-memory lifecycle facts the read surface consults. The client
983/// computes them from its own state; the module owns the decision so it can be
984/// tested with injected time and injected lifecycle state.
985#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
986pub struct PreviousVersionLifecycle {
987    /// §7.3.5 lease error or expiry.
988    pub lease_inactive: bool,
989    /// §3.3 scope revocation (any revoked subscription).
990    pub scope_revoked: bool,
991    /// §7.4.5 replacement coverage complete.
992    pub coverage_complete: bool,
993}
994
995/// D7 read spec.
996#[derive(Debug, Clone, Default, PartialEq, Eq)]
997pub struct PreviousVersionReadSpec {
998    pub table: String,
999    pub row_ids: Vec<String>,
1000    pub limit: Option<i64>,
1001}
1002
1003/// D7: `limit` defaults to 50, must be positive, and is capped at 200.
1004pub fn resolve_previous_version_limit(limit: Option<i64>) -> Result<i64, String> {
1005    match limit {
1006        None => Ok(PREVIOUS_VERSION_DEFAULT_LIMIT),
1007        Some(limit) if limit >= 1 => Ok(limit.min(PREVIOUS_VERSION_MAX_LIMIT)),
1008        Some(_) => Err(
1009            "sync.invalid_request: previousVersionSnapshot limit must be a positive number"
1010                .to_owned(),
1011        ),
1012    }
1013}
1014
1015/// Read the container's own metadata record, opening the sibling file for that
1016/// window only. `None` when there is no file or the record is unreadable.
1017pub fn read_previous_version_record(
1018    replica_path: &str,
1019) -> Result<Option<PreviousVersionRecord>, String> {
1020    let path = previous_version_container_path(replica_path);
1021    if !std::path::Path::new(&path).exists() {
1022        return Ok(None);
1023    }
1024    let container = Connection::open(&path)
1025        .map_err(|error| format!("open previous-version container {path:?}: {error}"))?;
1026    let record = read_previous_version_container(&container);
1027    let _ = container.close();
1028    // A container this module cannot read is treated as absent metadata, which
1029    // the boot reconcile discards on the next open.
1030    Ok(record.unwrap_or(None))
1031}
1032
1033/// D7: the read surface. `config` is the resolved feature config (`None` or
1034/// `enabled: false` ⇒ not configured); `lifecycle` is computed by the client
1035/// from in-memory state; `now_ms` is the injected clock.
1036///
1037/// Lease stop/expiry, scope revocation, coverage completion and the TTL all
1038/// DISCARD the container and close the read with their own reason — closing
1039/// reads without discarding would leave the plaintext behind.
1040pub fn previous_version_snapshot(
1041    replica: &Connection,
1042    replica_path: Option<&str>,
1043    config: Option<&PreviousVersionContextConfig>,
1044    lifecycle: PreviousVersionLifecycle,
1045    now_ms: i64,
1046    current_version: i32,
1047    spec: &PreviousVersionReadSpec,
1048) -> Result<PreviousVersionSnapshot, String> {
1049    let Some(config) = config.filter(|config| config.enabled) else {
1050        return Ok(PreviousVersionSnapshot::unavailable(
1051            current_version,
1052            PreviousVersionReason::NotConfigured,
1053        ));
1054    };
1055    let record = match replica_path {
1056        Some(replica_path) => read_previous_version_record(replica_path)?,
1057        None => None,
1058    };
1059    let discard = |reason: PreviousVersionReason| {
1060        if let Some(replica_path) = replica_path {
1061            discard_previous_version(replica, replica_path)?;
1062        }
1063        Ok(PreviousVersionSnapshot::unavailable(
1064            current_version,
1065            reason,
1066        ))
1067    };
1068    if lifecycle.lease_inactive {
1069        return discard(PreviousVersionReason::LeaseInactive);
1070    }
1071    if lifecycle.scope_revoked {
1072        return discard(PreviousVersionReason::ScopeRevoked);
1073    }
1074    if lifecycle.coverage_complete {
1075        return discard(PreviousVersionReason::CoverageComplete);
1076    }
1077    let Some(record) = record else {
1078        let reason = stored_previous_version_refusal(replica)
1079            .unwrap_or(PreviousVersionRefusalReason::NoPreviousDescriptor);
1080        return Ok(PreviousVersionSnapshot::unavailable(
1081            current_version,
1082            reason.as_read_reason(),
1083        ));
1084    };
1085    if record.current_version != current_version {
1086        return discard(PreviousVersionReason::NoPreviousDescriptor);
1087    }
1088    if now_ms - record.created_at_ms > config.max_age_ms {
1089        return discard(PreviousVersionReason::Expired);
1090    }
1091    let Some(table) = record.table(&spec.table) else {
1092        return Err(format!(
1093            "sync.invalid_request: previousVersionSnapshot names unknown previous table {:?}",
1094            spec.table
1095        ));
1096    };
1097    let Some(replica_path) = replica_path else {
1098        return Ok(PreviousVersionSnapshot::unavailable(
1099            current_version,
1100            PreviousVersionReason::NoPreviousDescriptor,
1101        ));
1102    };
1103    let path = previous_version_container_path(replica_path);
1104    let container = Connection::open(&path)
1105        .map_err(|error| format!("open previous-version container {path:?}: {error}"))?;
1106    let (rows, truncated) = read_previous_version_rows(
1107        &container,
1108        table,
1109        &spec.row_ids,
1110        resolve_previous_version_limit(spec.limit)?,
1111    )?;
1112    let _ = container.close();
1113    Ok(PreviousVersionSnapshot {
1114        state: PREVIOUS_VERSION_STATE,
1115        available: true,
1116        previous_version: Some(record.previous_version),
1117        current_version,
1118        reason: None,
1119        rows,
1120        truncated,
1121    })
1122}
1123
1124/// Decode the durable refusal; `None` when absent or corrupt. The successful
1125/// record lives inside the container file, so this key is absent whenever a
1126/// container is present.
1127pub fn stored_previous_version_refusal(
1128    replica: &Connection,
1129) -> Option<PreviousVersionRefusalReason> {
1130    let raw = meta_get(replica, PREVIOUS_VERSION_CONTEXT_KEY)?;
1131    let parsed: Value = serde_json::from_str(&raw).ok()?;
1132    let Value::Object(record) = &parsed else {
1133        return None;
1134    };
1135    if as_count(record.get("v")) != Some(1) {
1136        return None;
1137    }
1138    match record.get("reason").and_then(Value::as_str) {
1139        Some("no-previous-descriptor") => Some(PreviousVersionRefusalReason::NoPreviousDescriptor),
1140        Some("capture-exceeded-budget") => {
1141            Some(PreviousVersionRefusalReason::CaptureExceededBudget)
1142        }
1143        _ => None,
1144    }
1145}
1146
1147#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
1148#[serde(rename_all = "kebab-case")]
1149pub enum PreviousVersionAuditReason {
1150    UnknownTable,
1151    UnknownColumn,
1152}
1153
1154impl PreviousVersionAuditReason {
1155    #[must_use]
1156    pub fn name(self) -> &'static str {
1157        match self {
1158            Self::UnknownTable => "unknown-table",
1159            Self::UnknownColumn => "unknown-column",
1160        }
1161    }
1162}
1163
1164/// D6: one incompatible commit. Records the commit id, table, typed reason and
1165/// offending column ONLY — never an operation, row value or envelope field.
1166#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1167#[serde(rename_all = "camelCase")]
1168pub struct PreviousVersionAuditEntry {
1169    pub commit_id: String,
1170    pub table: String,
1171    pub reason: PreviousVersionAuditReason,
1172    #[serde(skip_serializing_if = "Option::is_none")]
1173    pub column: Option<String>,
1174}
1175
1176/// D6: the advisory pre-reset compatibility audit. Bounded at
1177/// [`MAX_AUDIT_INCOMPATIBLE`] entries.
1178#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
1179#[serde(rename_all = "camelCase")]
1180pub struct PreviousVersionAudit {
1181    pub v: i64,
1182    pub at_ms: i64,
1183    pub from_version: i32,
1184    pub to_version: i32,
1185    pub pending: i64,
1186    pub encodable: i64,
1187    pub truncated: bool,
1188    pub incompatible: Vec<PreviousVersionAuditEntry>,
1189}
1190
1191pub const MAX_AUDIT_INCOMPATIBLE: usize = 200;
1192
1193/// One pending outbox commit as the audit needs it: its id, and `(table, value
1194/// keys)` per operation.
1195#[derive(Debug, Clone, PartialEq, Eq)]
1196pub struct PendingCommitAudit<'a> {
1197    pub commit_id: &'a str,
1198    pub operations: Vec<(&'a str, Vec<&'a str>)>,
1199}
1200
1201/// The first reason these operations cannot re-encode under `schema`, mirroring
1202/// the TS `firstIncompatibility`. Shared with the §7.4.4 send-time drop so the
1203/// audit and the drop never disagree about what is incompatible.
1204pub fn first_incompatibility<'a>(
1205    schema: &ClientSchema,
1206    operations: &[(&'a str, Vec<&'a str>)],
1207) -> Option<(&'a str, PreviousVersionAuditReason, Option<&'a str>)> {
1208    for (table_name, value_keys) in operations {
1209        let Some(table) = schema.table(table_name) else {
1210            return Some((*table_name, PreviousVersionAuditReason::UnknownTable, None));
1211        };
1212        for key in value_keys {
1213            if !table.columns.iter().any(|column| column.name == *key) {
1214                return Some((
1215                    *table_name,
1216                    PreviousVersionAuditReason::UnknownColumn,
1217                    Some(*key),
1218                ));
1219            }
1220        }
1221    }
1222    None
1223}
1224
1225/// D6: classify pending outbox commits against the NEW compiled schema, before
1226/// the wipe. Advisory only — it drops nothing.
1227pub fn build_previous_version_audit(
1228    schema: &ClientSchema,
1229    pending: &[PendingCommitAudit<'_>],
1230    from_version: i32,
1231    to_version: i32,
1232    at_ms: i64,
1233) -> PreviousVersionAudit {
1234    let mut incompatible = Vec::new();
1235    let mut incompatible_total = 0usize;
1236    let mut encodable = 0i64;
1237    for commit in pending {
1238        let Some((table, reason, column)) = first_incompatibility(schema, &commit.operations)
1239        else {
1240            encodable += 1;
1241            continue;
1242        };
1243        incompatible_total += 1;
1244        if incompatible.len() < MAX_AUDIT_INCOMPATIBLE {
1245            incompatible.push(PreviousVersionAuditEntry {
1246                commit_id: commit.commit_id.to_owned(),
1247                table: table.to_owned(),
1248                reason,
1249                column: column.map(str::to_owned),
1250            });
1251        }
1252    }
1253    PreviousVersionAudit {
1254        v: 1,
1255        at_ms,
1256        from_version,
1257        to_version,
1258        pending: pending.len() as i64,
1259        encodable,
1260        truncated: incompatible_total > MAX_AUDIT_INCOMPATIBLE,
1261        incompatible,
1262    }
1263}
1264
1265pub fn write_previous_version_audit(replica: &Connection, audit: &PreviousVersionAudit) {
1266    if let Ok(encoded) = serde_json::to_string(audit) {
1267        meta_set(replica, PREVIOUS_VERSION_AUDIT_KEY, &encoded);
1268    }
1269}
1270
1271/// Strict decode; `None` when absent or malformed.
1272pub fn stored_previous_version_audit(replica: &Connection) -> Option<PreviousVersionAudit> {
1273    let raw = meta_get(replica, PREVIOUS_VERSION_AUDIT_KEY)?;
1274    let parsed: Value = serde_json::from_str(&raw).ok()?;
1275    let Value::Object(record) = &parsed else {
1276        return None;
1277    };
1278    if !has_exact_keys(
1279        record,
1280        &[
1281            "atMs",
1282            "encodable",
1283            "fromVersion",
1284            "incompatible",
1285            "pending",
1286            "toVersion",
1287            "truncated",
1288            "v",
1289        ],
1290    ) || as_count(record.get("v")) != Some(1)
1291        || record.get("truncated").and_then(Value::as_bool).is_none()
1292    {
1293        return None;
1294    }
1295    let count = |key: &str| as_count(record.get(key));
1296    let Some(Value::Array(entries)) = record.get("incompatible") else {
1297        return None;
1298    };
1299    let mut incompatible = Vec::with_capacity(entries.len());
1300    for entry in entries {
1301        let Value::Object(entry) = entry else {
1302            return None;
1303        };
1304        let (Some(Value::String(commit_id)), Some(Value::String(table))) =
1305            (entry.get("commitId"), entry.get("table"))
1306        else {
1307            return None;
1308        };
1309        let reason = match entry.get("reason").and_then(Value::as_str) {
1310            Some("unknown-table") => PreviousVersionAuditReason::UnknownTable,
1311            Some("unknown-column") => PreviousVersionAuditReason::UnknownColumn,
1312            _ => return None,
1313        };
1314        let column = match entry.get("column") {
1315            None | Some(Value::Null) => None,
1316            Some(Value::String(column)) => Some(column.clone()),
1317            Some(_) => return None,
1318        };
1319        incompatible.push(PreviousVersionAuditEntry {
1320            commit_id: commit_id.clone(),
1321            table: table.clone(),
1322            reason,
1323            column,
1324        });
1325    }
1326    Some(PreviousVersionAudit {
1327        v: 1,
1328        at_ms: count("atMs")?,
1329        from_version: i32::try_from(count("fromVersion")?).ok()?,
1330        to_version: i32::try_from(count("toVersion")?).ok()?,
1331        pending: count("pending")?,
1332        encodable: count("encodable")?,
1333        truncated: record.get("truncated").and_then(Value::as_bool)?,
1334        incompatible,
1335    })
1336}
1337
1338#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1339pub struct PreviousVersionDiscardOutcome {
1340    pub present: bool,
1341    pub discarded: bool,
1342}
1343
1344/// D9 boot hygiene: a container whose metadata is missing/undecodable, or
1345/// whose recorded `currentVersion` is not the running generated schema version,
1346/// is discarded before anything can read it.
1347///
1348/// This is the ONE gap the reset sweep cannot see. The capture commits in the
1349/// container's OWN transaction, independent of the replica's savepoint, so a
1350/// crash after that commit but before the replica commits leaves a container
1351/// whose `currentVersion` is the NEW schema beside the OLD marker. The next
1352/// open at the old schema version takes the same-version branch — no reset, no
1353/// sweep — and would otherwise serve that container. Idempotent: no container
1354/// file means no-op.
1355pub fn reconcile_previous_version_at_boot(
1356    replica: &Connection,
1357    replica_path: &str,
1358    current_version: i32,
1359    max_age_ms: Option<i64>,
1360    now_ms: i64,
1361) -> Result<(), String> {
1362    let path = previous_version_container_path(replica_path);
1363    if !std::path::Path::new(&path).exists() {
1364        return Ok(());
1365    }
1366    let record = match Connection::open(&path) {
1367        Ok(container) => {
1368            let record = read_previous_version_container(&container).unwrap_or(None);
1369            let _ = container.close();
1370            record
1371        }
1372        // Unreadable is the same decision as unknown metadata: discard it.
1373        Err(_) => None,
1374    };
1375    // D7/A1: the TTL is applied by an AWARE binary only (`max_age_ms` is set
1376    // when the feature is configured). It makes no claim about an unaware
1377    // same-schema rollback, which runs none of this code.
1378    let stale = match &record {
1379        // Missing or undecodable metadata: nothing can read it safely.
1380        None => true,
1381        Some(record) => {
1382            record.current_version != current_version
1383                || max_age_ms.is_some_and(|max_age_ms| now_ms - record.created_at_ms > max_age_ms)
1384        }
1385    };
1386    if !stale {
1387        return Ok(());
1388    }
1389    discard_previous_version(replica, replica_path)?;
1390    Ok(())
1391}
1392
1393/// D7/D9/A2: the executable discard — drop the container and both metadata
1394/// records and remove the FILE. Idempotent: a no-op succeeds, and a container
1395/// that cannot even be opened (torn file) is still removed and reported as
1396/// discarded rather than failing. This is the RFC 0006 key-loss contract: a
1397/// throw here would block reactivation while the plaintext is gone.
1398pub fn discard_previous_version(
1399    replica: &Connection,
1400    replica_path: &str,
1401) -> Result<PreviousVersionDiscardOutcome, String> {
1402    let path = previous_version_container_path(replica_path);
1403    let mut present = std::path::Path::new(&path).exists();
1404    if present {
1405        if let Ok(container) = Connection::open(&path) {
1406            let _ = drop_previous_version_container(&container);
1407            let _ = container.close();
1408        }
1409        remove_previous_version_files(&path);
1410    }
1411    if meta_get(replica, PREVIOUS_VERSION_CONTEXT_KEY).is_some()
1412        || meta_get(replica, PREVIOUS_VERSION_AUDIT_KEY).is_some()
1413    {
1414        present = true;
1415    }
1416    meta_delete(replica, PREVIOUS_VERSION_CONTEXT_KEY);
1417    meta_delete(replica, PREVIOUS_VERSION_AUDIT_KEY);
1418    Ok(PreviousVersionDiscardOutcome {
1419        present,
1420        discarded: present,
1421    })
1422}
1423#[cfg(test)]
1424mod tests {
1425    //! RFC 0005 phase-1 tests: descriptor round-trip and strict decode, the
1426    //! container in the sibling FILE with readable typed rows, every budget
1427    //! aborting with no file left, the no-descriptor refusal, and a discard
1428    //! that removes the PATH (read directly from storage, never through the
1429    //! feature's own read API).
1430
1431    use std::path::{Path, PathBuf};
1432
1433    use serde_json::json;
1434
1435    use super::*;
1436    use crate::api::{ClientLimits, Mutation};
1437    use crate::client::{SyncClient, LOCAL_SCHEMA_VERSION_KEY};
1438    use crate::schema::parse_schema_json;
1439
1440    fn enabled_config() -> PreviousVersionContextConfig {
1441        PreviousVersionContextConfig {
1442            enabled: true,
1443            ..Default::default()
1444        }
1445    }
1446
1447    #[test]
1448    fn previous_version_snapshot_reasons_use_injected_lifecycle_and_discard() {
1449        let temp = TempFiles::new("read-reasons");
1450        let replica_path = temp.replica().to_str().expect("utf-8 path").to_owned();
1451        seed_replica(temp.replica(), 2);
1452        let replica = Connection::open(temp.replica()).expect("open replica");
1453        let container_path = previous_version_container_path(&replica_path);
1454        let config = enabled_config();
1455        let spec = PreviousVersionReadSpec {
1456            table: "things".to_owned(),
1457            ..Default::default()
1458        };
1459        let lifecycle = PreviousVersionLifecycle::default();
1460
1461        // 1. Off and disabled are the same read outcome.
1462        for config in [None, Some(&PreviousVersionContextConfig::default())] {
1463            let snapshot = previous_version_snapshot(
1464                &replica,
1465                Some(&replica_path),
1466                config,
1467                lifecycle,
1468                0,
1469                2,
1470                &spec,
1471            )
1472            .expect("snapshot");
1473            assert!(!snapshot.available);
1474            assert_eq!(snapshot.reason, Some(PreviousVersionReason::NotConfigured));
1475            assert_eq!(snapshot.state, "previousVersion");
1476            assert_eq!(snapshot.current_version, 2);
1477            assert!(snapshot.previous_version.is_none());
1478            assert!(snapshot.rows.is_empty());
1479        }
1480
1481        // 2. A capture refused for want of a descriptor is a durable reason.
1482        let refused =
1483            capture_previous_version_from_replica(&replica, &replica_path, 3, 2, &config, 1_000)
1484                .expect("capture");
1485        assert!(matches!(
1486            refused,
1487            CaptureOutcome::Refused {
1488                reason: PreviousVersionRefusalReason::NoPreviousDescriptor,
1489                ..
1490            }
1491        ));
1492        let snapshot = previous_version_snapshot(
1493            &replica,
1494            Some(&replica_path),
1495            Some(&config),
1496            lifecycle,
1497            1_000,
1498            2,
1499            &spec,
1500        )
1501        .expect("snapshot");
1502        assert_eq!(
1503            snapshot.reason,
1504            Some(PreviousVersionReason::NoPreviousDescriptor)
1505        );
1506
1507        // 3. A budget refusal keeps its measured reason too.
1508        let tiny = PreviousVersionContextConfig {
1509            max_bytes: 1,
1510            enabled: true,
1511            ..Default::default()
1512        };
1513        capture_previous_version_from_replica(&replica, &replica_path, 1, 2, &tiny, 1_000)
1514            .expect("capture");
1515        let snapshot = previous_version_snapshot(
1516            &replica,
1517            Some(&replica_path),
1518            Some(&config),
1519            lifecycle,
1520            1_000,
1521            2,
1522            &spec,
1523        )
1524        .expect("snapshot");
1525        assert_eq!(
1526            snapshot.reason,
1527            Some(PreviousVersionReason::CaptureExceededBudget)
1528        );
1529        assert!(!Path::new(&container_path).exists());
1530
1531        // 4. A successful capture is readable with typed rows.
1532        let recapture = || {
1533            let outcome = capture_previous_version_from_replica(
1534                &replica,
1535                &replica_path,
1536                1,
1537                2,
1538                &config,
1539                1_000,
1540            )
1541            .expect("capture");
1542            assert!(matches!(outcome, CaptureOutcome::Captured(_)));
1543        };
1544        recapture();
1545        let snapshot = previous_version_snapshot(
1546            &replica,
1547            Some(&replica_path),
1548            Some(&config),
1549            lifecycle,
1550            1_000,
1551            2,
1552            &spec,
1553        )
1554        .expect("snapshot");
1555        assert!(snapshot.available);
1556        assert_eq!(snapshot.previous_version, Some(1));
1557        assert_eq!(snapshot.current_version, 2);
1558        assert_eq!(snapshot.reason, None);
1559        assert_eq!(snapshot.state, "previousVersion");
1560        assert!(!snapshot.truncated);
1561        assert_eq!(snapshot.rows.len(), 2);
1562        assert_eq!(snapshot.rows[0]["s"], json!("hello"));
1563        assert_eq!(snapshot.rows[0]["b"], json!(true));
1564        assert_eq!(snapshot.rows[0]["by"], json!({ "$bytes": "010203" }));
1565
1566        // Row selection, and an unknown previous table is a loud request error.
1567        let selected = PreviousVersionReadSpec {
1568            table: "things".to_owned(),
1569            row_ids: vec!["r2".to_owned()],
1570            limit: Some(1),
1571        };
1572        let snapshot = previous_version_snapshot(
1573            &replica,
1574            Some(&replica_path),
1575            Some(&config),
1576            lifecycle,
1577            1_000,
1578            2,
1579            &selected,
1580        )
1581        .expect("snapshot");
1582        assert_eq!(snapshot.rows.len(), 1);
1583        assert_eq!(snapshot.rows[0]["id"], json!("r2"));
1584        let unknown = PreviousVersionReadSpec {
1585            table: "absent".to_owned(),
1586            ..Default::default()
1587        };
1588        assert!(previous_version_snapshot(
1589            &replica,
1590            Some(&replica_path),
1591            Some(&config),
1592            lifecycle,
1593            1_000,
1594            2,
1595            &unknown
1596        )
1597        .expect_err("unknown previous table")
1598        .contains("unknown previous table"));
1599
1600        // 5. Lease stop/expiry, scope revocation and coverage completion each
1601        // DISCARD the container and close the read with their own reason.
1602        for (lifecycle, expected) in [
1603            (
1604                PreviousVersionLifecycle {
1605                    lease_inactive: true,
1606                    ..Default::default()
1607                },
1608                PreviousVersionReason::LeaseInactive,
1609            ),
1610            (
1611                PreviousVersionLifecycle {
1612                    scope_revoked: true,
1613                    ..Default::default()
1614                },
1615                PreviousVersionReason::ScopeRevoked,
1616            ),
1617            (
1618                PreviousVersionLifecycle {
1619                    coverage_complete: true,
1620                    ..Default::default()
1621                },
1622                PreviousVersionReason::CoverageComplete,
1623            ),
1624        ] {
1625            recapture();
1626            assert!(Path::new(&container_path).exists());
1627            let snapshot = previous_version_snapshot(
1628                &replica,
1629                Some(&replica_path),
1630                Some(&config),
1631                lifecycle,
1632                1_000,
1633                2,
1634                &spec,
1635            )
1636            .expect("snapshot");
1637            assert!(!snapshot.available, "{expected:?}");
1638            assert_eq!(snapshot.reason, Some(expected));
1639            assert!(snapshot.rows.is_empty());
1640            assert!(
1641                !Path::new(&container_path).exists(),
1642                "{expected:?} must physically remove the container"
1643            );
1644            assert_eq!(meta_get(&replica, PREVIOUS_VERSION_CONTEXT_KEY), None);
1645        }
1646    }
1647
1648    #[test]
1649    fn previous_version_ttl_applies_at_read_and_at_boot_with_injected_now() {
1650        let temp = TempFiles::new("ttl");
1651        let replica_path = temp.replica().to_str().expect("utf-8 path").to_owned();
1652        seed_replica(temp.replica(), 1);
1653        let replica = Connection::open(temp.replica()).expect("open replica");
1654        let container_path = previous_version_container_path(&replica_path);
1655        let config = enabled_config();
1656        let max_age_ms = config.max_age_ms;
1657        let spec = PreviousVersionReadSpec {
1658            table: "things".to_owned(),
1659            ..Default::default()
1660        };
1661        let lifecycle = PreviousVersionLifecycle::default();
1662        let capture_at = || {
1663            let outcome = capture_previous_version_from_replica(
1664                &replica,
1665                &replica_path,
1666                1,
1667                2,
1668                &config,
1669                1_000,
1670            )
1671            .expect("capture");
1672            assert!(matches!(outcome, CaptureOutcome::Captured(_)));
1673        };
1674
1675        // At the boundary the capture is still good; one millisecond later the
1676        // read expires it and removes the file.
1677        capture_at();
1678        let snapshot = previous_version_snapshot(
1679            &replica,
1680            Some(&replica_path),
1681            Some(&config),
1682            lifecycle,
1683            1_000 + max_age_ms,
1684            2,
1685            &spec,
1686        )
1687        .expect("snapshot");
1688        assert!(snapshot.available);
1689        let snapshot = previous_version_snapshot(
1690            &replica,
1691            Some(&replica_path),
1692            Some(&config),
1693            lifecycle,
1694            1_001 + max_age_ms,
1695            2,
1696            &spec,
1697        )
1698        .expect("snapshot");
1699        assert_eq!(snapshot.reason, Some(PreviousVersionReason::Expired));
1700        assert!(!Path::new(&container_path).exists());
1701
1702        // Boot: no TTL without an aware binary (`max_age_ms` is None when the
1703        // feature is off) — the container is NOT deleted by an unaware boot
1704        // path. This asserts the limitation: the TTL is an aware-binary
1705        // hygiene bound and nothing else.
1706        capture_at();
1707        reconcile_previous_version_at_boot(
1708            &replica,
1709            &replica_path,
1710            2,
1711            None,
1712            1_000 + max_age_ms * 10,
1713        )
1714        .expect("boot reconcile without a TTL");
1715        assert!(Path::new(&container_path).exists());
1716
1717        // Boot: aware, within the TTL → kept; beyond it → discarded.
1718        reconcile_previous_version_at_boot(
1719            &replica,
1720            &replica_path,
1721            2,
1722            Some(max_age_ms),
1723            1_000 + max_age_ms,
1724        )
1725        .expect("boot reconcile");
1726        assert!(Path::new(&container_path).exists());
1727        reconcile_previous_version_at_boot(
1728            &replica,
1729            &replica_path,
1730            2,
1731            Some(max_age_ms),
1732            1_001 + max_age_ms,
1733        )
1734        .expect("boot reconcile");
1735        assert!(!Path::new(&container_path).exists());
1736    }
1737
1738    #[test]
1739    fn previous_version_audit_records_typed_reasons_and_round_trips() {
1740        let temp = TempFiles::new("audit");
1741        seed_replica(temp.replica(), 1);
1742        let replica = Connection::open(temp.replica()).expect("open replica");
1743        let schema = schema(2);
1744
1745        let pending = vec![
1746            PendingCommitAudit {
1747                commit_id: "c1",
1748                operations: vec![("things", vec!["id", "removed"])],
1749            },
1750            PendingCommitAudit {
1751                commit_id: "c2",
1752                operations: vec![("gone", vec![])],
1753            },
1754            PendingCommitAudit {
1755                commit_id: "c3",
1756                operations: vec![("things", vec!["id"])],
1757            },
1758        ];
1759        let audit = build_previous_version_audit(&schema, &pending, 1, 2, 42);
1760        assert_eq!(audit.v, 1);
1761        assert_eq!(audit.at_ms, 42);
1762        assert_eq!(audit.from_version, 1);
1763        assert_eq!(audit.to_version, 2);
1764        assert_eq!(audit.pending, 3);
1765        assert_eq!(audit.encodable, 1);
1766        assert!(!audit.truncated);
1767        assert_eq!(audit.incompatible.len(), 2);
1768        assert_eq!(
1769            audit.incompatible[0],
1770            PreviousVersionAuditEntry {
1771                commit_id: "c1".to_owned(),
1772                table: "things".to_owned(),
1773                reason: PreviousVersionAuditReason::UnknownColumn,
1774                column: Some("removed".to_owned()),
1775            }
1776        );
1777        assert_eq!(
1778            audit.incompatible[1],
1779            PreviousVersionAuditEntry {
1780                commit_id: "c2".to_owned(),
1781                table: "gone".to_owned(),
1782                reason: PreviousVersionAuditReason::UnknownTable,
1783                column: None,
1784            }
1785        );
1786        // The record carries a typed reason and NO envelope, operation or value.
1787        let encoded = serde_json::to_string(&audit).expect("encode audit");
1788        for forbidden in ["operations", "values", "payload", "envelope", "row_id"] {
1789            assert!(!encoded.contains(forbidden), "{encoded}");
1790        }
1791        assert!(encoded.contains("\"reason\":\"unknown-column\""));
1792        assert!(encoded.contains("\"reason\":\"unknown-table\""));
1793
1794        write_previous_version_audit(&replica, &audit);
1795        assert_eq!(stored_previous_version_audit(&replica), Some(audit.clone()));
1796        // Strict decode: a malformed record reads as absent.
1797        meta_set(&replica, PREVIOUS_VERSION_AUDIT_KEY, "{\"v\":1}");
1798        assert_eq!(stored_previous_version_audit(&replica), None);
1799        meta_set(
1800            &replica,
1801            PREVIOUS_VERSION_AUDIT_KEY,
1802            "{\"v\":1,\"atMs\":0,\"fromVersion\":1,\"toVersion\":2,\"pending\":0,\"encodable\":0,\"truncated\":false,\"incompatible\":[{\"commitId\":\"c\",\"table\":\"t\",\"reason\":\"other\"}]}",
1803        );
1804        assert_eq!(stored_previous_version_audit(&replica), None);
1805
1806        // Bounded at MAX_AUDIT_INCOMPATIBLE entries with `truncated`.
1807        let many = (0..MAX_AUDIT_INCOMPATIBLE + 5)
1808            .map(|_index| PendingCommitAudit {
1809                commit_id: "c",
1810                operations: vec![("gone", vec![])],
1811            })
1812            .collect::<Vec<_>>();
1813        let audit = build_previous_version_audit(&schema, &many, 1, 2, 0);
1814        assert_eq!(audit.pending, MAX_AUDIT_INCOMPATIBLE as i64 + 5);
1815        assert_eq!(audit.encodable, 0);
1816        assert_eq!(audit.incompatible.len(), MAX_AUDIT_INCOMPATIBLE);
1817        assert!(audit.truncated);
1818    }
1819
1820    #[test]
1821    fn previous_version_limit_defaults_and_ceiling_match_ts() {
1822        assert_eq!(resolve_previous_version_limit(None), Ok(50));
1823        assert_eq!(resolve_previous_version_limit(Some(7)), Ok(7));
1824        assert_eq!(resolve_previous_version_limit(Some(999)), Ok(200));
1825        assert!(resolve_previous_version_limit(Some(0)).is_err());
1826        assert!(resolve_previous_version_limit(Some(-1)).is_err());
1827    }
1828
1829    /// Removes every temp file the test created, so the container assertion
1830    /// "the path is gone" cannot pass because a later case reused the name.
1831    struct TempFiles {
1832        paths: Vec<PathBuf>,
1833    }
1834
1835    impl TempFiles {
1836        fn new(label: &str) -> Self {
1837            let path = std::env::temp_dir().join(format!(
1838                "syncular-prev-context-{label}-{}",
1839                uuid::Uuid::new_v4()
1840            ));
1841            let paths = vec![
1842                path.clone(),
1843                PathBuf::from(previous_version_container_path(
1844                    path.to_str().expect("utf-8 temp path"),
1845                )),
1846            ];
1847            Self { paths }
1848        }
1849
1850        fn replica(&self) -> &Path {
1851            &self.paths[0]
1852        }
1853    }
1854
1855    impl Drop for TempFiles {
1856        fn drop(&mut self) {
1857            for path in &self.paths {
1858                let _ = std::fs::remove_file(path);
1859                let _ = std::fs::remove_file(format!("{}-wal", path.display()));
1860                let _ = std::fs::remove_file(format!("{}-shm", path.display()));
1861            }
1862        }
1863    }
1864
1865    fn schema_json(version: i32) -> Value {
1866        json!({
1867            "version": version,
1868            "tables": [{
1869                "name": "things",
1870                "primaryKey": "id",
1871                "columns": [
1872                    { "name": "id", "type": "string", "nullable": false },
1873                    { "name": "project_id", "type": "string", "nullable": false },
1874                    { "name": "s", "type": "string", "nullable": true },
1875                    { "name": "i", "type": "integer", "nullable": true },
1876                    { "name": "f", "type": "float", "nullable": true },
1877                    { "name": "b", "type": "boolean", "nullable": true },
1878                    { "name": "j", "type": "json", "nullable": true },
1879                    { "name": "by", "type": "bytes", "nullable": true },
1880                    { "name": "cr", "type": "crdt", "nullable": true },
1881                    { "name": "br", "type": "blob_ref", "nullable": true },
1882                    { "name": "meta", "type": "string", "nullable": true }
1883                ],
1884                "scopes": []
1885            }, {
1886                "name": "notes",
1887                "primaryKey": "id",
1888                "columns": [
1889                    { "name": "id", "type": "string", "nullable": false },
1890                    { "name": "body", "type": "string", "nullable": true }
1891                ],
1892                "scopes": []
1893            }]
1894        })
1895    }
1896
1897    fn schema(version: i32) -> ClientSchema {
1898        parse_schema_json(&schema_json(version)).expect("valid test schema")
1899    }
1900
1901    /// The local DDL the client writes: bare column names, no declared type.
1902    fn seed_replica(path: &Path, rows: i64) {
1903        let replica = Connection::open(path).expect("open replica");
1904        replica
1905            .execute_batch(
1906                "CREATE TABLE _syncular_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
1907                 CREATE TABLE things (id, project_id, s, i, f, b, j, by, cr, br, meta);
1908                 CREATE TABLE notes (id, body);",
1909            )
1910            .expect("create local tables");
1911        for index in 0..rows {
1912            replica
1913                .execute(
1914                    "INSERT INTO things(rowid, id, project_id, s, i, f, b, j, by, cr, br, meta)
1915                       VALUES (?1, ?2, 'p1', 'hello', 7, 1.5, 1, '{\"a\":1}',
1916                               x'010203', x'0908', 'blob:xyz', 'm1')",
1917                    rusqlite::params![index + 1, format!("r{}", index + 1)],
1918                )
1919                .expect("insert row");
1920        }
1921        replica
1922            .execute("INSERT INTO notes(id, body) VALUES ('n1', 'note')", [])
1923            .expect("insert note");
1924        // D1: the descriptor the bump's capture reads, exactly as the client
1925        // writes it — plus the §7.4.1 marker of the version that wrote it.
1926        set_local_schema_descriptor(&replica, &schema(1)).expect("descriptor");
1927        meta_set(&replica, LOCAL_SCHEMA_VERSION_KEY, "1");
1928    }
1929
1930    fn descriptor() -> LocalSchemaDescriptor {
1931        build_local_schema_descriptor(&schema(1))
1932    }
1933
1934    #[test]
1935    fn previous_version_descriptor_round_trips_and_rejects_unknown_shapes() {
1936        let conn = Connection::open_in_memory().expect("open in-memory");
1937        conn.execute_batch(
1938            "CREATE TABLE _syncular_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)",
1939        )
1940        .expect("create meta");
1941        let schema = schema(4);
1942        set_local_schema_descriptor(&conn, &schema).expect("descriptor");
1943
1944        let stored = meta_get(&conn, LOCAL_SCHEMA_DESCRIPTOR_KEY).expect("descriptor written");
1945        let parsed: Value = serde_json::from_str(&stored).expect("descriptor is JSON");
1946        assert_eq!(parsed.as_object().expect("object").len(), 3, "{stored}");
1947        assert_eq!(parsed["v"], json!(1));
1948        assert_eq!(parsed["version"], json!(4));
1949        let tables = parsed["tables"].as_array().expect("tables array");
1950        assert_eq!(tables.len(), 2);
1951        assert_eq!(tables[0]["name"], json!("things"));
1952        assert_eq!(tables[0]["primaryKey"], json!("id"));
1953        assert_eq!(
1954            tables[0]["columns"][1],
1955            json!({ "name": "project_id", "type": "string" })
1956        );
1957        assert_eq!(
1958            tables[0]["columns"][5],
1959            json!({ "name": "b", "type": "boolean" })
1960        );
1961        assert_eq!(
1962            tables[0]["columns"][7],
1963            json!({ "name": "by", "type": "bytes" })
1964        );
1965
1966        // Round trip through the strict decoder.
1967        assert_eq!(
1968            decode_local_schema_descriptor(&stored).expect("strict decode"),
1969            build_local_schema_descriptor(&schema)
1970        );
1971        assert_eq!(
1972            load_local_schema_descriptor(&conn).expect("loaded"),
1973            build_local_schema_descriptor(&schema)
1974        );
1975
1976        // Strict decode: an unknown shape is corruption, never a best guess.
1977        for raw in [
1978            r#"{"v":1,"version":4,"tables":[],"extra":1}"#,
1979            r#"{"v":2,"version":4,"tables":[]}"#,
1980            r#"{"v":1,"version":4}"#,
1981            r#"{"v":1,"version":-1,"tables":[]}"#,
1982            r#"{"v":1,"version":4,"tables":{}}"#,
1983            r#"{"v":1,"version":4,"tables":[{"name":"t","primaryKey":"id"}]}"#,
1984            r#"{"v":1,"version":4,"tables":[{"name":"t","primaryKey":"id","columns":[{"name":"a","type":"nope"}]}]}"#,
1985            r#"{"v":1,"version":4,"tables":[{"name":"t","primaryKey":"id","columns":[]}],"x":0}"#,
1986            "not json",
1987        ] {
1988            assert!(
1989                decode_local_schema_descriptor(raw).is_err(),
1990                "must reject {raw}"
1991            );
1992        }
1993
1994        // A corrupt stored value reads as absent (the reset then refuses).
1995        meta_set(&conn, LOCAL_SCHEMA_DESCRIPTOR_KEY, "{broken");
1996        assert_eq!(load_local_schema_descriptor(&conn), None);
1997    }
1998
1999    #[test]
2000    fn previous_version_capture_writes_sibling_container_with_readable_rows() {
2001        let temp = TempFiles::new("capture");
2002        let replica_path = temp.replica().to_str().expect("utf-8 path").to_owned();
2003        seed_replica(temp.replica(), 2);
2004        let replica = Connection::open(temp.replica()).expect("open replica");
2005
2006        let outcome = capture_previous_version_from_replica(
2007            &replica,
2008            &replica_path,
2009            1,
2010            2,
2011            &PreviousVersionContextConfig::default(),
2012            1_700_000_000_000,
2013        )
2014        .expect("capture");
2015        let CaptureOutcome::Captured(measurement) = outcome else {
2016            panic!("expected a capture, got {outcome:?}");
2017        };
2018        assert_eq!(measurement.tables, 2);
2019        assert_eq!(measurement.rows, 3);
2020        assert!(measurement.bytes > 0);
2021
2022        // The container is a SEPARATE file, and the replica cannot see it.
2023        let container_path = previous_version_container_path(&replica_path);
2024        assert!(Path::new(&container_path).exists());
2025        let visible: i64 = replica
2026            .query_row(
2027                "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?1",
2028                rusqlite::params![CONTAINER_TABLE],
2029                |row| row.get(0),
2030            )
2031            .expect("count replica tables");
2032        assert_eq!(visible, 0);
2033
2034        let container = Connection::open(&container_path).expect("open container");
2035        let record = read_previous_version_container(&container)
2036            .expect("read record")
2037            .expect("record present");
2038        assert_eq!(record.previous_version, 1);
2039        assert_eq!(record.current_version, 2);
2040        assert_eq!(record.rows, 3);
2041        assert_eq!(record.bytes, measurement.bytes);
2042        assert_eq!(record.created_at_ms, 1_700_000_000_000);
2043        assert_eq!(record.tables.len(), 2);
2044        assert_eq!(record.tables[0].columns.len(), 11);
2045
2046        let table = record.table("things").expect("things in the record");
2047        let (rows, truncated) =
2048            read_previous_version_rows(&container, table, &[], 50).expect("read rows");
2049        assert!(!truncated);
2050        assert_eq!(rows.len(), 2);
2051        let row = &rows[0];
2052        assert_eq!(row["id"], json!("r1"));
2053        assert_eq!(row["s"], json!("hello"));
2054        assert_eq!(row["i"], json!(7));
2055        assert_eq!(row["f"], json!(1.5));
2056        assert_eq!(row["b"], json!(true));
2057        assert_eq!(row["j"], json!("{\"a\":1}"));
2058        assert_eq!(row["br"], json!("blob:xyz"));
2059        assert_eq!(row["by"], json!({ "$bytes": "010203" }));
2060        assert_eq!(row["cr"], json!({ "$bytes": "0908" }));
2061        assert!(!row.contains_key("_syncular_version"));
2062        assert!(!row.contains_key("_sync_version"));
2063
2064        // Row selection and the `limit + 1` truncation probe.
2065        let (selected, _) =
2066            read_previous_version_rows(&container, table, &["r2".to_owned()], 50).expect("select");
2067        assert_eq!(selected.len(), 1);
2068        assert_eq!(selected[0]["id"], json!("r2"));
2069        let (missing, _) =
2070            read_previous_version_rows(&container, table, &["absent".to_owned()], 50)
2071                .expect("select missing");
2072        assert!(missing.is_empty());
2073        let (limited, truncated) =
2074            read_previous_version_rows(&container, table, &[], 0).expect("limit 0");
2075        assert!(limited.is_empty());
2076        assert!(truncated);
2077    }
2078
2079    #[test]
2080    fn previous_version_each_budget_refuses_and_leaves_no_file() {
2081        let temp = TempFiles::new("budget");
2082        let replica_path = temp.replica().to_str().expect("utf-8 path").to_owned();
2083        seed_replica(temp.replica(), 2);
2084        let replica = Connection::open(temp.replica()).expect("open replica");
2085        let container_path = previous_version_container_path(&replica_path);
2086
2087        let cases = [
2088            (
2089                "maxTables",
2090                PreviousVersionContextConfig {
2091                    max_tables: 1,
2092                    ..Default::default()
2093                },
2094                0,
2095                false,
2096            ),
2097            (
2098                "maxRows",
2099                PreviousVersionContextConfig {
2100                    max_rows: 1,
2101                    ..Default::default()
2102                },
2103                2,
2104                false,
2105            ),
2106            (
2107                "maxBytes",
2108                PreviousVersionContextConfig {
2109                    max_bytes: 1,
2110                    ..Default::default()
2111                },
2112                3,
2113                true,
2114            ),
2115            (
2116                "maxRowBytes",
2117                PreviousVersionContextConfig {
2118                    max_row_bytes: 1,
2119                    ..Default::default()
2120                },
2121                3,
2122                false,
2123            ),
2124        ];
2125        for (label, config, expected_rows, bytes_measured) in cases {
2126            let outcome =
2127                capture_previous_version_from_replica(&replica, &replica_path, 1, 2, &config, 1)
2128                    .unwrap_or_else(|error| panic!("{label}: {error}"));
2129            let CaptureOutcome::Refused {
2130                reason,
2131                measurement,
2132            } = outcome
2133            else {
2134                panic!("{label}: expected a refusal, got {outcome:?}");
2135            };
2136            assert_eq!(
2137                reason,
2138                PreviousVersionRefusalReason::CaptureExceededBudget,
2139                "{label}"
2140            );
2141            assert_eq!(measurement.tables, 2, "{label}");
2142            assert_eq!(measurement.rows, expected_rows, "{label}");
2143            if bytes_measured {
2144                assert!(measurement.bytes > 1, "{label}");
2145            } else {
2146                assert_eq!(measurement.bytes, 0, "{label}");
2147            }
2148            // All-or-nothing: no partial copy, and no file at all.
2149            assert!(!Path::new(&container_path).exists(), "{label}");
2150            let refusal = meta_get(&replica, PREVIOUS_VERSION_CONTEXT_KEY)
2151                .unwrap_or_else(|| panic!("{label}: refusal recorded"));
2152            assert!(refusal.contains("capture-exceeded-budget"), "{refusal}");
2153        }
2154    }
2155
2156    #[test]
2157    fn previous_version_missing_or_stale_descriptor_refuses_without_a_file() {
2158        let temp = TempFiles::new("no-descriptor");
2159        let replica_path = temp.replica().to_str().expect("utf-8 path").to_owned();
2160        seed_replica(temp.replica(), 1);
2161        let replica = Connection::open(temp.replica()).expect("open replica");
2162        let container_path = previous_version_container_path(&replica_path);
2163
2164        // No descriptor at all: a database last opened by an unaware binary.
2165        meta_delete(&replica, LOCAL_SCHEMA_DESCRIPTOR_KEY);
2166        let outcome = capture_previous_version_from_replica(
2167            &replica,
2168            &replica_path,
2169            1,
2170            2,
2171            &PreviousVersionContextConfig::default(),
2172            1,
2173        )
2174        .expect("capture");
2175        assert_eq!(
2176            outcome,
2177            CaptureOutcome::Refused {
2178                reason: PreviousVersionRefusalReason::NoPreviousDescriptor,
2179                measurement: CaptureMeasurement::default(),
2180            }
2181        );
2182        assert!(!Path::new(&container_path).exists());
2183        let refusal = meta_get(&replica, PREVIOUS_VERSION_CONTEXT_KEY).expect("refusal");
2184        assert!(refusal.contains("no-previous-descriptor"), "{refusal}");
2185
2186        // A descriptor for a DIFFERENT version is treated exactly the same.
2187        meta_set(
2188            &replica,
2189            LOCAL_SCHEMA_DESCRIPTOR_KEY,
2190            &encode_local_schema_descriptor(&descriptor()),
2191        );
2192        let outcome = capture_previous_version_from_replica(
2193            &replica,
2194            &replica_path,
2195            3,
2196            4,
2197            &PreviousVersionContextConfig::default(),
2198            1,
2199        )
2200        .expect("capture");
2201        assert!(matches!(
2202            outcome,
2203            CaptureOutcome::Refused {
2204                reason: PreviousVersionRefusalReason::NoPreviousDescriptor,
2205                ..
2206            }
2207        ));
2208        assert!(!Path::new(&container_path).exists());
2209    }
2210
2211    #[test]
2212    fn previous_version_discard_removes_the_container_path() {
2213        let temp = TempFiles::new("discard");
2214        let replica_path = temp.replica().to_str().expect("utf-8 path").to_owned();
2215        seed_replica(temp.replica(), 1);
2216        let replica = Connection::open(temp.replica()).expect("open replica");
2217        let container_path = previous_version_container_path(&replica_path);
2218
2219        let outcome = capture_previous_version_from_replica(
2220            &replica,
2221            &replica_path,
2222            1,
2223            2,
2224            &PreviousVersionContextConfig::default(),
2225            1,
2226        )
2227        .expect("capture");
2228        assert!(matches!(outcome, CaptureOutcome::Captured(_)));
2229        assert!(Path::new(&container_path).exists());
2230        meta_set(&replica, PREVIOUS_VERSION_AUDIT_KEY, "{\"v\":1}");
2231
2232        let discarded = discard_previous_version(&replica, &replica_path).expect("discard");
2233        assert!(discarded.present);
2234        assert!(discarded.discarded);
2235        // The PATH is gone — not merely the tables inside it.
2236        assert!(!Path::new(&container_path).exists());
2237        assert!(!Path::new(&format!("{container_path}-wal")).exists());
2238        assert!(!Path::new(&format!("{container_path}-shm")).exists());
2239        assert_eq!(meta_get(&replica, PREVIOUS_VERSION_CONTEXT_KEY), None);
2240        assert_eq!(meta_get(&replica, PREVIOUS_VERSION_AUDIT_KEY), None);
2241
2242        // A no-op discard succeeds (the RFC 0006 key-loss contract needs it).
2243        let again = discard_previous_version(&replica, &replica_path).expect("discard again");
2244        assert!(!again.present);
2245        assert!(!again.discarded);
2246    }
2247
2248    #[test]
2249    fn previous_version_torn_container_never_blocks_sweep_discard_or_open() {
2250        let temp = TempFiles::new("torn");
2251        let replica_path = temp.replica().to_str().expect("utf-8 path").to_owned();
2252        let container_path = previous_version_container_path(&replica_path);
2253
2254        // FIX 1: garbage at the sibling path makes the connection open but
2255        // every statement fail with "file is not a database".
2256        std::fs::write(&container_path, b"not a sqlite database").expect("write garbage");
2257        let replica = Connection::open(temp.replica()).expect("open replica");
2258
2259        // The sweep must still remove it, and must not error.
2260        sweep_previous_version_container(&replica, &replica_path)
2261            .expect("a torn container must not fail the sweep");
2262        assert!(!Path::new(&container_path).exists());
2263
2264        // The discard must report it and still remove it.
2265        std::fs::write(&container_path, b"not a sqlite database").expect("write garbage again");
2266        let discarded = discard_previous_version(&replica, &replica_path)
2267            .expect("a torn container must not fail the discard");
2268        assert!(discarded.present);
2269        assert!(discarded.discarded);
2270        assert!(!Path::new(&container_path).exists());
2271
2272        // And a client opening over a torn container must still open, sweeping
2273        // it on the bump path.
2274        std::fs::write(&container_path, b"not a sqlite database")
2275            .expect("write garbage a third time");
2276        seed_replica(temp.replica(), 1);
2277        {
2278            SyncClient::open_path(
2279                "pvc".to_owned(),
2280                &schema_json(2),
2281                ClientLimits::default(),
2282                &replica_path,
2283            )
2284            .expect("a torn container must not block the client open");
2285        }
2286        assert!(!Path::new(&container_path).exists());
2287    }
2288
2289    #[test]
2290    fn previous_version_same_version_open_discards_a_container_it_did_not_bump_to() {
2291        // Reproduces the exact cross-file interleaving: the replica savepoint
2292        // opens (marker still 1), the sweep runs, the container's OWN
2293        // transaction commits {previousVersion: 1, currentVersion: 2}, and the
2294        // process dies before the replica savepoint releases. The next open is
2295        // at the OLD schema version, so it runs no reset and no sweep.
2296        let temp = TempFiles::new("same-version-orphan");
2297        let replica_path = temp.replica().to_str().expect("utf-8 path").to_owned();
2298        seed_replica(temp.replica(), 1);
2299        let replica = Connection::open(temp.replica()).expect("open replica");
2300        let container_path = previous_version_container_path(&replica_path);
2301
2302        let outcome = capture_previous_version_from_replica(
2303            &replica,
2304            &replica_path,
2305            1,
2306            2,
2307            &PreviousVersionContextConfig::default(),
2308            1,
2309        )
2310        .expect("capture");
2311        assert!(matches!(outcome, CaptureOutcome::Captured(_)));
2312        assert!(Path::new(&container_path).exists());
2313        // The replica never committed: the marker is still 1.
2314        assert_eq!(
2315            meta_get(&replica, LOCAL_SCHEMA_VERSION_KEY).as_deref(),
2316            Some("1")
2317        );
2318        drop(replica);
2319
2320        // Opening at the SAME version (1) must discard the stale container
2321        // before any read surface can see it, and open normally.
2322        {
2323            let client = SyncClient::open_path(
2324                "pvc".to_owned(),
2325                &schema_json(1),
2326                ClientLimits::default(),
2327                &replica_path,
2328            )
2329            .expect("the old-schema open must succeed");
2330            let rows = client
2331                .query("SELECT COUNT(*) AS c FROM things", &[])
2332                .expect("the client still serves ordinary queries");
2333            assert_eq!(rows.len(), 1);
2334        }
2335        // Read storage directly, never through a read API.
2336        assert!(!Path::new(&container_path).exists());
2337        let inspect = Connection::open(&replica_path).expect("inspect after open");
2338        assert_eq!(meta_get(&inspect, PREVIOUS_VERSION_CONTEXT_KEY), None);
2339        assert_eq!(meta_get(&inspect, PREVIOUS_VERSION_AUDIT_KEY), None);
2340        assert_eq!(
2341            inspect
2342                .query_row(
2343                    "SELECT COUNT(*) FROM sqlite_master WHERE name = ?1",
2344                    rusqlite::params![CONTAINER_TABLE],
2345                    |row| row.get::<_, i64>(0),
2346                )
2347                .expect("replica table count"),
2348            0
2349        );
2350
2351        // A container whose recorded currentVersion MATCHES the running schema
2352        // is left alone (it is the container this boot would serve).
2353        let temp = TempFiles::new("same-version-live");
2354        let replica_path = temp.replica().to_str().expect("utf-8 path").to_owned();
2355        seed_replica(temp.replica(), 1);
2356        let replica = Connection::open(temp.replica()).expect("open replica");
2357        capture_previous_version_from_replica(
2358            &replica,
2359            &replica_path,
2360            1,
2361            1,
2362            &PreviousVersionContextConfig::default(),
2363            1,
2364        )
2365        .expect("capture");
2366        drop(replica);
2367        let container_path = previous_version_container_path(&replica_path);
2368        assert!(Path::new(&container_path).exists());
2369        SyncClient::open_path(
2370            "pvc".to_owned(),
2371            &schema_json(1),
2372            ClientLimits::default(),
2373            &replica_path,
2374        )
2375        .expect("open v1 beside a matching container");
2376        assert!(Path::new(&container_path).exists());
2377    }
2378
2379    #[test]
2380    fn previous_version_client_writes_descriptor_and_captures_on_bump() {
2381        let temp = TempFiles::new("client");
2382        let replica_path = temp.replica().to_str().expect("utf-8 path").to_owned();
2383        let container_path = previous_version_container_path(&replica_path);
2384
2385        // Fresh install at v1: the descriptor is written beside the marker and
2386        // the feature (default off) creates no container.
2387        {
2388            let mut client = SyncClient::open_path(
2389                "pvc".to_owned(),
2390                &schema_json(1),
2391                ClientLimits::default(),
2392                &replica_path,
2393            )
2394            .expect("open v1");
2395            client
2396                .mutate(vec![Mutation::Upsert {
2397                    table: "things".to_owned(),
2398                    values: Map::from_iter([
2399                        ("id".to_owned(), json!("r1")),
2400                        ("project_id".to_owned(), json!("p1")),
2401                        ("s".to_owned(), json!("hello")),
2402                    ]),
2403                    base_version: None,
2404                }])
2405                .expect("seed a local row");
2406        }
2407        {
2408            let inspect = Connection::open(&replica_path).expect("inspect v1");
2409            let descriptor =
2410                load_local_schema_descriptor(&inspect).expect("descriptor after install");
2411            assert_eq!(descriptor.version, 1);
2412            assert_eq!(descriptor.tables.len(), 2);
2413            assert_eq!(
2414                descriptor.table("things").expect("things").columns.len(),
2415                11
2416            );
2417        }
2418        assert!(!Path::new(&container_path).exists());
2419
2420        // A same-version open backfills a descriptor an unaware binary never wrote.
2421        {
2422            let inspect = Connection::open(&replica_path).expect("inspect");
2423            meta_delete(&inspect, LOCAL_SCHEMA_DESCRIPTOR_KEY);
2424        }
2425        {
2426            let _client = SyncClient::open_path(
2427                "pvc".to_owned(),
2428                &schema_json(1),
2429                ClientLimits::default(),
2430                &replica_path,
2431            )
2432            .expect("reopen v1");
2433        }
2434        {
2435            let inspect = Connection::open(&replica_path).expect("inspect v1 again");
2436            assert_eq!(
2437                load_local_schema_descriptor(&inspect)
2438                    .expect("backfilled")
2439                    .version,
2440                1
2441            );
2442        }
2443
2444        // The bump captures the v1 row into the sibling file, and the replica
2445        // connection still cannot see the container table.
2446        {
2447            let limits = ClientLimits {
2448                previous_version_context: Some(PreviousVersionContextConfig {
2449                    enabled: true,
2450                    ..Default::default()
2451                }),
2452                ..Default::default()
2453            };
2454            SyncClient::open_path("pvc".to_owned(), &schema_json(2), limits, &replica_path)
2455                .expect("bump to v2");
2456        }
2457        assert!(Path::new(&container_path).exists());
2458        {
2459            let inspect = Connection::open(&replica_path).expect("inspect v2");
2460            assert_eq!(
2461                load_local_schema_descriptor(&inspect)
2462                    .expect("descriptor after bump")
2463                    .version,
2464                2
2465            );
2466            assert_eq!(
2467                inspect
2468                    .query_row(
2469                        "SELECT COUNT(*) FROM sqlite_master WHERE name = ?1",
2470                        rusqlite::params![CONTAINER_TABLE],
2471                        |row| row.get::<_, i64>(0),
2472                    )
2473                    .expect("replica table count"),
2474                0
2475            );
2476        }
2477        {
2478            let container = Connection::open(&container_path).expect("open container");
2479            let record = read_previous_version_container(&container)
2480                .expect("read record")
2481                .expect("captured record");
2482            assert_eq!(record.previous_version, 1);
2483            assert_eq!(record.current_version, 2);
2484            assert_eq!(record.rows, 1);
2485            let (rows, _) = read_previous_version_rows(
2486                &container,
2487                record.table("things").expect("things"),
2488                &[],
2489                50,
2490            )
2491            .expect("read rows");
2492            assert_eq!(rows.len(), 1);
2493            assert_eq!(rows[0]["s"], json!("hello"));
2494        }
2495
2496        // Discard removes the path and both metadata records.
2497        {
2498            let inspect = Connection::open(&replica_path).expect("inspect v2");
2499            let outcome = discard_previous_version(&inspect, &replica_path).expect("discard");
2500            assert!(outcome.present && outcome.discarded);
2501        }
2502        assert!(!Path::new(&container_path).exists());
2503    }
2504}