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