1use 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
39pub const LOCAL_SCHEMA_DESCRIPTOR_KEY: &str = "localSchemaDescriptor";
41pub const PREVIOUS_VERSION_CONTEXT_KEY: &str = "previousVersionContext";
45pub const PREVIOUS_VERSION_AUDIT_KEY: &str = "previousVersionAudit";
48
49pub const PREVIOUS_VERSION_CONTAINER_SUFFIX: &str = ".prev-context";
52
53const CONTAINER_TABLE: &str = "syncular_prev_context";
55const CONTAINER_META_TABLE: &str = "_syncular_prev_context_meta";
57
58const DESCRIPTOR_VERSION: i64 = 1;
59const RECORD_VERSION: i64 = 1;
60const COPY_BATCH_ROWS: i64 = 500;
62
63#[derive(Debug, Clone, Copy, PartialEq, Eq)]
65pub struct PreviousVersionContextConfig {
66 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 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 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
114const 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
122pub fn previous_version_container_path(replica_path: &str) -> String {
125 format!("{replica_path}{PREVIOUS_VERSION_CONTAINER_SUFFIX}")
126}
127
128#[derive(Debug, Clone, PartialEq, Eq)]
133pub struct DescriptorColumn {
134 pub name: String,
135 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
159pub 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
208fn 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
214fn 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
264pub 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
291pub 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
302pub 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
308pub 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#[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 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
428pub 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
453pub 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#[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
575fn 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
600pub 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 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 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 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 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
803fn 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
827fn remove_previous_version_files(path: &str) {
834 for suffix in ["", "-wal", "-shm"] {
835 let _ = std::fs::remove_file(format!("{path}{suffix}"));
836 }
837}
838
839pub 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
864pub 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 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
926pub const PREVIOUS_VERSION_DEFAULT_LIMIT: i64 = 50;
928pub const PREVIOUS_VERSION_MAX_LIMIT: i64 = 200;
929
930#[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
960pub const PREVIOUS_VERSION_STATE: &str = "previousVersion";
962
963#[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#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
980pub struct PreviousVersionLifecycle {
981 pub lease_inactive: bool,
983 pub scope_revoked: bool,
985 pub coverage_complete: bool,
987}
988
989#[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
997pub 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
1009pub 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 Ok(record.unwrap_or(None))
1025}
1026
1027pub 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
1115pub 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#[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#[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#[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
1192pub 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
1216pub 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
1262pub 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
1335pub 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 Err(_) => None,
1365 };
1366 let stale = match &record {
1370 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
1384pub 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 assert_eq!(
2306 meta_get(&replica, LOCAL_SCHEMA_VERSION_KEY).as_deref(),
2307 Some("1")
2308 );
2309 drop(replica);
2310
2311 {
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 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 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 {
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 {
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 {
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 {
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}