feat: compress sqlite db (#2278)

<!-- Please make sure there is an issue that this PR is correlated to. -->

## Changes

<!-- If there are frontend changes, please include screenshots. -->

Author: MasterPtato

packages/common/chirp-workflow/core/src/db/fdb_sqlite_nats/mod.rs

@@ -205,18 +205,7 @@ impl Database for DatabaseFdbSqliteNats {
 	fn from_pools(pools: rivet_pools::Pools) -> Result<Arc<Self>, rivet_pools::Error> {
 		// Start background flush handler task
 		let (flush_tx, flush_rx) = mpsc::unbounded_channel();
-		let pools2 = pools.clone();
-		tokio::spawn(async move {
-			if let Err(err) = flush_handler(pools2, flush_rx).await {
-				tracing::error!(
-					?err,
-					"fdb_sqlite_nats workflow driver flush handler task failed"
-				);
-
-				// Shut down entire runtime
-				rivet_runtime::shutdown().await;
-			}
-		});
+		tokio::spawn(flush_handler(pools.clone(), flush_rx));
 
 		Ok(Arc::new(DatabaseFdbSqliteNats {
 			pools,
@@ -3237,24 +3226,24 @@ enum WorkflowState {
 	Silenced,
 }
 
-async fn flush_handler(
-	pools: rivet_pools::Pools,
-	mut flush_rx: mpsc::UnboundedReceiver<Uuid>,
-) -> WorkflowResult<()> {
+async fn flush_handler(pools: rivet_pools::Pools, mut flush_rx: mpsc::UnboundedReceiver<Uuid>) {
 	while let Some(workflow_id) = flush_rx.recv().await {
 		tracing::debug!(?workflow_id, "flushing workflow");
 
-		pools
+		if let Err(err) = pools
 			.sqlite_manager()
 			.flush(vec![
 				sqlite::db_name_internal(workflow_id),
 				crate::db::sqlite_db_name_data(workflow_id),
 			])
-			.await?;
+			.await
+		{
+			// TODO: Somehow forward the error to the workflow so it can die
+			tracing::error!(?workflow_id, ?err, "failed to flush workflow databases");
+		}
 	}
 
 	// If the channel is closed that means the db driver instance was dropped which is not an error
-	Ok(())
 }
 
 fn value_to_str(v: &serde_json::Value) -> WorkflowResult<String> {

packages/common/fdb-util/src/keys.rs

@@ -45,6 +45,7 @@ pub const CLIENTS_BY_REMAINING_MEM: usize = 43;
 pub const SQLITE: usize = 44;
 pub const INTERNAL: usize = 45;
 pub const METADATA: usize = 46;
+pub const COMPRESSED_DATA: usize = 47;
 
 // Directories with fdbrs must use string paths instead of tuples
 pub mod dir {

packages/common/pools/Cargo.toml

@@ -20,6 +20,7 @@ hex.workspace = true
 hyper = { version = "0.14" }
 hyper-tls = { version = "0.5.0" }
 lazy_static = "1.4"
+lz4_flex = "0.11.3"
 rivet-config.workspace = true
 rivet-metrics.workspace = true
 service-discovery.workspace = true

packages/common/pools/src/db/sqlite/keys.rs

@@ -11,26 +11,6 @@ impl DbDataKey {
 	}
 }
 
-//impl FormalChunkedKey for SqliteDbKey {
-//	type ChunkKey = SqliteDbKey;
-//	type Value = Vec<u8>;
-//
-//	fn chunk(&self, chunk: usize) -> Self::ChunkKey {
-//		SqliteDbKey {
-//			workflow_id: self.workflow_id,
-//			chunk,
-//		}
-//	}
-//
-//	fn combine(&self, chunks: Vec<FdbValue>) -> Result<Self::Value> {
-//		TODO
-//	}
-//
-//	fn split(&self, value: Self::Value) -> Result<Vec<Vec<u8>>> {
-//		self.split_ref(value.as_ref())
-//	}
-//}
-
 impl TuplePack for DbDataKey {
 	fn pack<W: std::io::Write>(
 		&self,
@@ -42,27 +22,68 @@ impl TuplePack for DbDataKey {
 	}
 }
 
+/// Uncompressed data.
 pub struct DbDataChunkKey {
+	#[allow(dead_code)]
+	pub db_name_segment: Arc<Vec<u8>>,
+	pub chunk: usize,
+}
+
+impl<'de> TupleUnpack<'de> for DbDataChunkKey {
+	fn unpack(input: &[u8], tuple_depth: TupleDepth) -> PackResult<(&[u8], Self)> {
+		let (input, (_, db_name_segment, _, chunk)) =
+			<(usize, Vec<u8>, usize, usize)>::unpack(input, tuple_depth)?;
+		let v = DbDataChunkKey {
+			db_name_segment: Arc::new(db_name_segment),
+			chunk,
+		};
+
+		Ok((input, v))
+	}
+}
+
+pub struct CompressedDbDataKey {
+	db_name_segment: Arc<Vec<u8>>,
+}
+
+impl CompressedDbDataKey {
+	pub fn new(db_name_segment: Arc<Vec<u8>>) -> Self {
+		CompressedDbDataKey { db_name_segment }
+	}
+}
+
+impl TuplePack for CompressedDbDataKey {
+	fn pack<W: std::io::Write>(
+		&self,
+		w: &mut W,
+		tuple_depth: TupleDepth,
+	) -> std::io::Result<VersionstampOffset> {
+		let t = (DBS, &*self.db_name_segment, COMPRESSED_DATA);
+		t.pack(w, tuple_depth)
+	}
+}
+
+pub struct CompressedDbDataChunkKey {
 	pub db_name_segment: Arc<Vec<u8>>,
 	pub chunk: usize,
 }
 
-impl TuplePack for DbDataChunkKey {
+impl TuplePack for CompressedDbDataChunkKey {
 	fn pack<W: std::io::Write>(
 		&self,
 		w: &mut W,
 		tuple_depth: TupleDepth,
 	) -> std::io::Result<VersionstampOffset> {
-		let t = (DBS, &*self.db_name_segment, DATA, self.chunk);
+		let t = (DBS, &*self.db_name_segment, COMPRESSED_DATA, self.chunk);
 		t.pack(w, tuple_depth)
 	}
 }
 
-impl<'de> TupleUnpack<'de> for DbDataChunkKey {
+impl<'de> TupleUnpack<'de> for CompressedDbDataChunkKey {
 	fn unpack(input: &[u8], tuple_depth: TupleDepth) -> PackResult<(&[u8], Self)> {
 		let (input, (_, db_name_segment, _, chunk)) =
 			<(usize, Vec<u8>, usize, usize)>::unpack(input, tuple_depth)?;
-		let v = DbDataChunkKey {
+		let v = CompressedDbDataChunkKey {
 			db_name_segment: Arc::new(db_name_segment),
 			chunk,
 		};