assetpack_core/sqlite/rusqlite/
gc.rs

1use std::collections::{BTreeSet, HashSet};
2
3use super::{Error, Hash32, MerkleMode, ObjectKind, ObjectSource, Result, RusqliteStore, parse_recipe_checked};
4
5#[derive(Debug, Clone, Default, PartialEq, Eq)]
6pub struct GcReport {
7  pub deleted_chunks: u64,
8  pub deleted_recipes: u64,
9  pub freed_bytes: u64,
10}
11
12#[derive(Debug, Clone, Default, PartialEq, Eq)]
13pub struct ClosureReport {
14  pub missing_objects: Vec<Hash32>,
15  pub orphan_chunks: u64,
16  pub orphan_recipes: u64,
17}
18
19impl RusqliteStore<'_> {
20  /// Deletes objects and dangling file recipe cache entries in the caller's
21  /// transaction.
22  pub fn delete_objects(&self, hashes: &[Hash32]) -> Result<()> {
23    self.gc_savepoint(|| {
24      let sql = format!("DELETE FROM {} WHERE hash=?1", self.layout.objects);
25      let mut statement = self.connection.prepare(&sql)?;
26      for hash in hashes {
27        statement.execute([hash.as_bytes().as_slice()])?;
28      }
29      self.finish_object_deletion()
30    })
31  }
32
33  /// Collects objects outside the live recipe closure in the caller's
34  /// transaction. All live recipes and chunk references are checked before
35  /// any object is deleted.
36  pub fn collect_garbage(&self, live_recipes: &[Hash32]) -> Result<GcReport> {
37    self.gc_savepoint(|| {
38      let (live, missing) = self.recipe_closure(live_recipes)?;
39      if !missing.is_empty() {
40        return Err(Error::Integrity(format!(
41          "live recipe closure is missing {} objects",
42          missing.len()
43        )));
44      }
45      self
46        .connection
47        .execute_batch("CREATE TEMP TABLE assetpack_gc_live (hash BLOB PRIMARY KEY) WITHOUT ROWID;")?;
48      {
49        let mut insert = self.connection.prepare("INSERT INTO temp.assetpack_gc_live (hash) VALUES (?1)")?;
50        for hash in live {
51          insert.execute([hash.as_bytes().as_slice()])?;
52        }
53      }
54      let orphan = format!(
55        "NOT EXISTS (SELECT 1 FROM temp.assetpack_gc_live live WHERE live.hash={}.hash)",
56        self.layout.objects
57      );
58      let sql = format!(
59        "SELECT kind,COUNT(*),SUM(stored_size) FROM {} WHERE {orphan} GROUP BY kind",
60        self.layout.objects
61      );
62      let mut statement = self.connection.prepare(&sql)?;
63      let mut report = GcReport::default();
64      for row in statement.query_map([], |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?, row.get::<_, i64>(2)?)))? {
65        let (kind, count, bytes) = row?;
66        let count = u64::try_from(count).map_err(|_| Error::Integrity("negative orphan count".into()))?;
67        let bytes = u64::try_from(bytes).map_err(|_| Error::Integrity("negative orphan stored size".into()))?;
68        match ObjectKind::from_i64(kind) {
69          Some(ObjectKind::Chunk) => report.deleted_chunks += count,
70          Some(ObjectKind::Recipe) => report.deleted_recipes += count,
71          None => return Err(Error::Integrity("invalid object kind during garbage collection".into())),
72        }
73        report.freed_bytes += bytes;
74      }
75      self
76        .connection
77        .execute(&format!("DELETE FROM {} WHERE {orphan}", self.layout.objects), [])?;
78      self.finish_object_deletion()?;
79      self.connection.execute("DROP TABLE temp.assetpack_gc_live", [])?;
80      Ok(report)
81    })
82  }
83
84  /// Reads recipes and object metadata only; this also works on read-only
85  /// connections.
86  pub fn check_closure(&self, live_recipes: &[Hash32]) -> Result<ClosureReport> {
87    let (live, missing_objects) = self.recipe_closure(live_recipes)?;
88    let mut report = ClosureReport {
89      missing_objects,
90      ..Default::default()
91    };
92    let mut statement = self.connection.prepare(&format!("SELECT hash,kind FROM {}", self.layout.objects))?;
93    for row in statement.query_map([], |row| Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, i64>(1)?)))? {
94      let (hash, kind) = row?;
95      if live.contains(&Hash32::from_bytes(&hash)?) {
96        continue;
97      }
98      match ObjectKind::from_i64(kind) {
99        Some(ObjectKind::Chunk) => report.orphan_chunks += 1,
100        Some(ObjectKind::Recipe) => report.orphan_recipes += 1,
101        None => return Err(Error::Integrity("invalid object kind during closure check".into())),
102      }
103    }
104    Ok(report)
105  }
106
107  fn recipe_closure(&self, live_recipes: &[Hash32]) -> Result<(HashSet<Hash32>, Vec<Hash32>)> {
108    let mut live = HashSet::new();
109    let mut missing = BTreeSet::new();
110    let mut chunks = HashSet::new();
111    for hash in live_recipes.iter().copied().collect::<BTreeSet<_>>() {
112      live.insert(hash);
113      let Some(object) = self.read_object(&hash)? else {
114        missing.insert(hash);
115        continue;
116      };
117      if object.kind != ObjectKind::Recipe {
118        return Err(Error::ObjectKindMismatch {
119          hash,
120          expected: ObjectKind::Recipe,
121          actual: object.kind,
122        });
123      }
124      for (chunk, _) in parse_recipe_checked(&object.bytes, &hash)?.chunks {
125        live.insert(chunk);
126        chunks.insert(chunk);
127      }
128    }
129    let mut statement = self
130      .connection
131      .prepare(&format!("SELECT kind FROM {} WHERE hash=?1", self.layout.objects))?;
132    for hash in chunks {
133      use rusqlite::OptionalExtension;
134      let kind = statement
135        .query_row([hash.as_bytes().as_slice()], |row| row.get::<_, i64>(0))
136        .optional()?;
137      match kind {
138        None => {
139          missing.insert(hash);
140        }
141        Some(kind) if kind == ObjectKind::Chunk as i64 => {}
142        Some(kind) => {
143          let actual = ObjectKind::from_i64(kind).ok_or_else(|| Error::Integrity("invalid chunk object kind".into()))?;
144          return Err(Error::ObjectKindMismatch {
145            hash,
146            expected: ObjectKind::Chunk,
147            actual,
148          });
149        }
150      }
151    }
152    Ok((live, missing.into_iter().collect()))
153  }
154
155  fn gc_savepoint<T>(&self, action: impl FnOnce() -> Result<T>) -> Result<T> {
156    if self.connection.is_autocommit() {
157      return Err(Error::InvalidConfig("object deletion requires a caller transaction".into()));
158    }
159    self.connection.execute_batch("SAVEPOINT assetpack_object_deletion")?;
160    match action() {
161      Ok(value) => {
162        self.connection.execute_batch("RELEASE assetpack_object_deletion")?;
163        Ok(value)
164      }
165      Err(error) => {
166        self
167          .connection
168          .execute_batch("ROLLBACK TO assetpack_object_deletion; RELEASE assetpack_object_deletion")?;
169        Err(error)
170      }
171    }
172  }
173
174  fn finish_object_deletion(&self) -> Result<()> {
175    self.connection.execute(
176      &format!(
177        "DELETE FROM {} WHERE NOT EXISTS (SELECT 1 FROM {} objects WHERE objects.hash=recipe_hash AND objects.kind={})",
178        self.layout.file_recipe_cache,
179        self.layout.objects,
180        ObjectKind::Recipe as i64,
181      ),
182      [],
183    )?;
184    if self.merkle == MerkleMode::Enabled {
185      self.rebuild_merkle_index()?;
186    }
187    Ok(())
188  }
189}
190
191#[cfg(test)]
192mod tests {
193  use rusqlite::Connection;
194
195  use super::{GcReport, RusqliteStore};
196  use crate::{Codec, Hash32, MerkleMode, ObjectSource, SqliteStoreOptions, build_recipe};
197
198  fn store(connection: &Connection) -> RusqliteStore<'_> {
199    RusqliteStore::from_connection_with_options(
200      connection,
201      SqliteStoreOptions {
202        namespace: Some("test".into()),
203        merkle: MerkleMode::Disabled,
204      },
205    )
206    .unwrap()
207  }
208
209  fn recipe(store: &RusqliteStore<'_>, content: &[u8], chunk: Hash32) -> Hash32 {
210    let file = Hash32::sha3_256(content);
211    let bytes = build_recipe(content.len() as u64, &[(chunk, content.len() as u32)], file, 0, 0);
212    let hash = Hash32::sha3_256(&bytes);
213    store.put_recipe(hash, &bytes, Codec::Raw).unwrap();
214    store.put_file_recipe_cache_batch(&[(file, hash)]).unwrap();
215    hash
216  }
217
218  #[test]
219  fn garbage_collection_preserves_shared_chunks_and_cleans_recipe_cache() {
220    let mut connection = Connection::open_in_memory().unwrap();
221    let blobs = store(&connection);
222    let shared = Hash32::sha3_256(b"shared");
223    let orphan = Hash32::sha3_256(b"orphan");
224    blobs.put_chunk(shared, b"shared", Codec::Raw).unwrap();
225    blobs.put_chunk(orphan, b"orphan", Codec::Raw).unwrap();
226    let live = recipe(&blobs, b"live", shared);
227    let dead = recipe(&blobs, b"dead", shared);
228    let closure = blobs.check_closure(&[live]).unwrap();
229    assert!(closure.missing_objects.is_empty());
230    assert_eq!((closure.orphan_chunks, closure.orphan_recipes), (1, 1));
231    let dead_bytes = blobs.read_object(&dead).unwrap().unwrap().bytes.len() as u64;
232    let transaction = connection.transaction().unwrap();
233    let report = store(&transaction).collect_garbage(&[live, live]).unwrap();
234    assert_eq!(
235      report,
236      GcReport {
237        deleted_chunks: 1,
238        deleted_recipes: 1,
239        freed_bytes: 6 + dead_bytes
240      }
241    );
242    transaction.commit().unwrap();
243    let blobs = store(&connection);
244    assert!(blobs.read_object(&shared).unwrap().is_some());
245    assert!(blobs.read_object(&dead).unwrap().is_none());
246    assert_eq!(blobs.recipe_for_file_hash(&Hash32::sha3_256(b"dead")).unwrap(), None);
247    assert_eq!(blobs.recipe_for_file_hash(&Hash32::sha3_256(b"live")).unwrap(), Some(live));
248    let transaction = connection.transaction().unwrap();
249    store(&transaction).delete_objects(&[live]).unwrap();
250    transaction.rollback().unwrap();
251    assert_eq!(store(&connection).file_recipe_cache().unwrap().len(), 1);
252    let transaction = connection.transaction().unwrap();
253    store(&transaction).delete_objects(&[live]).unwrap();
254    transaction.commit().unwrap();
255    assert!(store(&connection).file_recipe_cache().unwrap().is_empty());
256  }
257
258  #[test]
259  fn invalid_live_closure_never_deletes_objects_even_if_caller_commits() {
260    for invalid in ["missing_recipe", "malformed_recipe", "missing_chunk", "wrong_kind"] {
261      let mut connection = Connection::open_in_memory().unwrap();
262      let blobs = store(&connection);
263      let orphan = Hash32::sha3_256(b"keep-on-failure");
264      blobs.put_chunk(orphan, b"keep-on-failure", Codec::Raw).unwrap();
265      let live = match invalid {
266        "missing_recipe" => Hash32::sha3_256(b"missing"),
267        "malformed_recipe" => {
268          let hash = Hash32::sha3_256(b"malformed");
269          blobs.put_recipe(hash, b"malformed", Codec::Raw).unwrap();
270          hash
271        }
272        "missing_chunk" => recipe(&blobs, b"live", Hash32::sha3_256(b"absent-chunk")),
273        "wrong_kind" => orphan,
274        _ => unreachable!(),
275      };
276      let before = blobs.verified_object_hashes().unwrap();
277      let cache = blobs.file_recipe_cache().unwrap();
278      let transaction = connection.transaction().unwrap();
279      assert!(store(&transaction).collect_garbage(&[live]).is_err(), "{invalid}");
280      transaction.commit().unwrap();
281      assert_eq!(store(&connection).verified_object_hashes().unwrap(), before, "{invalid}");
282      assert_eq!(store(&connection).file_recipe_cache().unwrap(), cache, "{invalid}");
283    }
284  }
285
286  #[test]
287  fn closure_check_is_read_only_and_gc_joins_host_rollback() {
288    let mut connection = Connection::open_in_memory().unwrap();
289    let blobs = store(&connection);
290    let chunk = Hash32::sha3_256(b"chunk");
291    blobs.put_chunk(chunk, b"chunk", Codec::Raw).unwrap();
292    let live = recipe(&blobs, b"live", chunk);
293    let missing = Hash32::sha3_256(b"missing");
294    connection.pragma_update(None, "query_only", true).unwrap();
295    let read_only = RusqliteStore::from_existing_connection_with_options(
296      &connection,
297      SqliteStoreOptions {
298        namespace: Some("test".into()),
299        merkle: MerkleMode::Disabled,
300      },
301    )
302    .unwrap();
303    assert_eq!(read_only.check_closure(&[live, missing]).unwrap().missing_objects, vec![missing]);
304    connection.pragma_update(None, "query_only", false).unwrap();
305    assert!(blobs.collect_garbage(&[]).is_err());
306    assert!(blobs.delete_objects(&[chunk]).is_err());
307    let transaction = connection.transaction().unwrap();
308    assert_eq!(store(&transaction).collect_garbage(&[]).unwrap().deleted_chunks, 1);
309    transaction.rollback().unwrap();
310    assert!(store(&connection).read_object(&chunk).unwrap().is_some());
311    let transaction = connection.transaction().unwrap();
312    store(&transaction).collect_garbage(&[]).unwrap();
313    transaction.commit().unwrap();
314    assert!(store(&connection).verified_object_hashes().unwrap().is_empty());
315    assert!(store(&connection).file_recipe_cache().unwrap().is_empty());
316  }
317
318  #[test]
319  fn gc_failure_rolls_back_sweep_and_preserves_foreign_temp_table() {
320    let mut connection = Connection::open_in_memory().unwrap();
321    let blobs = store(&connection);
322    let chunk = Hash32::sha3_256(b"chunk");
323    blobs.put_chunk(chunk, b"chunk", Codec::Raw).unwrap();
324    recipe(&blobs, b"live", chunk);
325    connection
326      .execute_batch("CREATE TEMP TABLE assetpack_gc_live (marker TEXT); INSERT INTO assetpack_gc_live VALUES ('caller');")
327      .unwrap();
328    let transaction = connection.transaction().unwrap();
329    assert!(store(&transaction).collect_garbage(&[]).is_err());
330    transaction.commit().unwrap();
331    assert_eq!(
332      connection
333        .query_row("SELECT marker FROM temp.assetpack_gc_live", [], |row| row.get::<_, String>(0))
334        .unwrap(),
335      "caller"
336    );
337    connection
338      .execute_batch(
339        "DROP TABLE temp.assetpack_gc_live; CREATE TRIGGER fail_cache_delete BEFORE DELETE ON test_file_recipe_cache BEGIN SELECT \
340         RAISE(ABORT, 'injected'); END;",
341      )
342      .unwrap();
343    let before = store(&connection).verified_object_hashes().unwrap();
344    let transaction = connection.transaction().unwrap();
345    assert!(store(&transaction).collect_garbage(&[]).is_err());
346    transaction.commit().unwrap();
347    assert_eq!(store(&connection).verified_object_hashes().unwrap(), before);
348    connection.execute_batch("DROP TRIGGER fail_cache_delete").unwrap();
349    let transaction = connection.transaction().unwrap();
350    store(&transaction).collect_garbage(&[]).unwrap();
351    transaction.commit().unwrap();
352    assert!(store(&connection).verified_object_hashes().unwrap().is_empty());
353  }
354
355  #[test]
356  fn deletion_rebuilds_enabled_merkle_index_in_host_transaction() {
357    let mut connection = Connection::open_in_memory().unwrap();
358    let blobs = RusqliteStore::from_connection(&connection).unwrap();
359    let chunk = Hash32::sha3_256(b"chunk");
360    blobs.put_chunk(chunk, b"chunk", Codec::Raw).unwrap();
361    let live = recipe(&blobs, b"live", chunk);
362    let before = blobs.rebuild_merkle_index().unwrap();
363    let transaction = connection.transaction().unwrap();
364    let blobs = RusqliteStore::from_connection(&transaction).unwrap();
365    blobs.collect_garbage(&[]).unwrap();
366    assert_eq!(blobs.merkle_summary().unwrap().leaf_count, 0);
367    transaction.rollback().unwrap();
368    assert_eq!(
369      RusqliteStore::from_connection(&connection).unwrap().merkle_summary().unwrap(),
370      before
371    );
372    let transaction = connection.transaction().unwrap();
373    let blobs = RusqliteStore::from_connection(&transaction).unwrap();
374    blobs.delete_objects(&[live]).unwrap();
375    assert_eq!(blobs.merkle_summary().unwrap(), blobs.recompute_merkle_summary().unwrap());
376    assert_eq!(blobs.merkle_summary().unwrap().leaf_count, 1);
377    transaction.commit().unwrap();
378  }
379}
assetpack_core/sqlite/rusqlite/gc.rs

assetpack_core/sqlite/rusqlite/
gc.rs
