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