feat: Implement func run and func run log layerdb fallbacks on the new client

Author: vbustamante

Cargo.lock

@@ -506,6 +506,15 @@ impl SiDbContext for DalContext {
     fn change_set_id(&self) -> ChangeSetId {
         self.change_set_id()
     }
+
+    // TODO get rid of these after we don't need layer db fallbacks
+    fn func_run_layer_db(&self) -> &si_layer_cache::db::func_run::FuncRunLayerDb {
+        (&self.services_context.layer_db).func_run()
+    }
+
+    fn func_run_log_layer_db(&self) -> &si_layer_cache::db::func_run_log::FuncRunLogLayerDb {
+        (&self.services_context.layer_db).func_run_log()
+    }
 }
 
 impl SiDbTransactions for Transactions {

lib/dal/src/context.rs

@@ -11,6 +11,7 @@ rust_library(
         "//lib/si-data-pg:si-data-pg",
         "//lib/si-events-rs:si-events",
         "//lib/si-id:si-id",
+        "//lib/si-layer-cache:si-layer-cache",
         "//lib/telemetry-rs:telemetry",
         "//third-party/rust:async-trait",
         "//third-party/rust:chrono",

lib/si-db/BUCK

@@ -23,6 +23,7 @@ si-data-nats = { path = "../../lib/si-data-nats" }
 si-data-pg = { path = "../../lib/si-data-pg" }
 si-events = { path = "../../lib/si-events-rs" }
 si-id = { path = "../../lib/si-id" }
+si-layer-cache = { path = "../../lib/si-layer-cache" }
 strum = { workspace = true }
 telemetry = { path = "../../lib/telemetry-rs" }
 thiserror = { workspace = true }

lib/si-db/Cargo.toml

@@ -1,5 +1,9 @@
 use async_trait::async_trait;
 use si_id::ChangeSetId;
+use si_layer_cache::db::{
+    func_run::FuncRunLayerDb,
+    func_run_log::FuncRunLogLayerDb,
+};
 use tokio::sync::MappedMutexGuard;
 
 use crate::{
@@ -21,4 +25,7 @@ pub trait SiDbContext {
     fn tenancy(&self) -> &Tenancy;
     fn visibility(&self) -> &Visibility;
     fn change_set_id(&self) -> ChangeSetId;
+    // TODO get rid of these after we don't need layer db fallbacks
+    fn func_run_layer_db(&self) -> &FuncRunLayerDb;
+    fn func_run_log_layer_db(&self) -> &FuncRunLogLayerDb;
 }

lib/si-db/src/context.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use si_events::{
     ActionId,
     AttributeValueId,
@@ -84,6 +86,7 @@ impl FuncRunDb {
         let postcard_bytes =
             postcard::to_stdvec(&func_run).map_err(|e| SiDbError::Postcard(e.to_string()))?;
 
+        // Write to si-db
         ctx.txns()
             .await?
             .pg()
@@ -147,9 +150,21 @@ impl FuncRunDb {
             )
             .await?;
 
+        // Also write to layer-db for backward compatibility during migration
+        ctx.func_run_layer_db()
+            .write(
+                Arc::new(func_run.clone()),
+                None, // web_events
+                *func_run.tenancy(),
+                *func_run.actor(),
+            )
+            .await
+            .map_err(|e| SiDbError::LayerDb(e.to_string()))?;
+
         Ok(())
     }
 
+    // NOTE(victor): This is only used by migration so it does not write to layer-db
     /// Write multiple func runs to the database in a single INSERT query.
     /// This is more efficient than calling upsert multiple times.
     pub async fn upsert_batch(ctx: &impl SiDbContext, func_runs: Vec<FuncRun>) -> SiDbResult<()> {
@@ -270,6 +285,7 @@ impl FuncRunDb {
         Ok(())
     }
 
+    // NOTE(victor): This is only used by migration so it does not fallback to layer-db
     /// Returns the IDs from the input batch that do NOT exist in the database.
     /// This is useful for determining which func runs need to be migrated.
     pub async fn find_missing_ids(
@@ -323,7 +339,11 @@ impl FuncRunDb {
                 .map_err(|e| SiDbError::Postcard(e.to_string()))?;
             Ok(Some(func_run))
         } else {
-            Ok(None)
+            // Fall back to layer-db if not found in si-db
+            ctx.func_run_layer_db()
+                .get_last_run_for_action_id_opt(workspace_pk, action_id)
+                .await
+                .map_err(|e| SiDbError::LayerDb(e.to_string()))
         }
     }
 
@@ -360,6 +380,18 @@ impl FuncRunDb {
             func_runs.push(func_run);
         }
 
+        // If si-db returned empty, fall back to layer-db
+        if func_runs.is_empty() {
+            if let Some(layer_func_runs) = ctx
+                .func_run_layer_db()
+                .list_management_history(workspace_pk, change_set_id)
+                .await
+                .map_err(|e| SiDbError::LayerDb(e.to_string()))?
+            {
+                return Ok(layer_func_runs);
+            }
+        }
+
         Ok(func_runs)
     }
 
@@ -386,7 +418,16 @@ impl FuncRunDb {
                 .map_err(|e| SiDbError::Postcard(e.to_string()))?;
             Ok(Some(func_run))
         } else {
-            Ok(None)
+            // Fall back to layer-db if not found in si-db
+            ctx.func_run_layer_db()
+                .get_last_management_run_for_func_and_component_id(
+                    workspace_pk,
+                    change_set_id,
+                    component_id,
+                    func_id,
+                )
+                .await
+                .map_err(|e| SiDbError::LayerDb(e.to_string()))
         }
     }
 
@@ -411,7 +452,11 @@ impl FuncRunDb {
                 .map_err(|e| SiDbError::Postcard(e.to_string()))?;
             Ok(Some(func_run))
         } else {
-            Ok(None)
+            // Fall back to layer-db if not found in si-db
+            ctx.func_run_layer_db()
+                .get_last_qualification_for_attribute_value_id(workspace_pk, attribute_value_id)
+                .await
+                .map_err(|e| SiDbError::LayerDb(e.to_string()))
         }
     }
 
@@ -432,29 +477,19 @@ impl FuncRunDb {
                 .map_err(|e| SiDbError::Postcard(e.to_string()))?;
             Ok(Some(func_run))
         } else {
-            Ok(None)
+            // Fall back to layer-db if not found in si-db
+            ctx.func_run_layer_db()
+                .try_read(key)
+                .await
+                .map(|arc_func_run| Some((*arc_func_run).clone()))
+                .map_err(|e| SiDbError::LayerDb(e.to_string()))
         }
     }
 
     pub async fn try_read(ctx: &impl SiDbContext, key: FuncRunId) -> SiDbResult<FuncRun> {
-        let maybe_row = ctx
-            .txns()
+        Self::read(ctx, key)
             .await?
-            .pg()
-            .query_opt(
-                &format!("SELECT value FROM {DBNAME} WHERE key = $1"),
-                &[&key.to_string()],
-            )
-            .await?;
-
-        if let Some(row) = maybe_row {
-            let value_bytes: Vec<u8> = row.try_get("value")?;
-            let func_run: FuncRun = postcard::from_bytes(&value_bytes)
-                .map_err(|e| SiDbError::Postcard(e.to_string()))?;
-            Ok(func_run)
-        } else {
-            Err(SiDbError::MissingFuncRun(key))
-        }
+            .ok_or(SiDbError::MissingFuncRun(key))
     }
 
     pub async fn read_many_for_workspace(
@@ -476,6 +511,18 @@ impl FuncRunDb {
             func_runs.push(func_run);
         }
 
+        // Fall back to layer-db if si-db returned no results
+        if func_runs.is_empty() {
+            if let Some(layer_func_runs) = ctx
+                .func_run_layer_db()
+                .read_many_for_workspace(workspace_pk)
+                .await
+                .map_err(|e| SiDbError::LayerDb(e.to_string()))?
+            {
+                return Ok(layer_func_runs.into_iter().map(|arc| (*arc).clone()).collect());
+            }
+        }
+
         Ok(func_runs)
     }
 
@@ -527,6 +574,18 @@ impl FuncRunDb {
             func_runs.push(func_run);
         }
 
+        // Fall back to layer-db if si-db returned no results
+        if func_runs.is_empty() {
+            if let Some(layer_func_runs) = ctx
+                .func_run_layer_db()
+                .read_many_for_workspace_paginated(workspace_pk, change_set_id, limit, cursor)
+                .await
+                .map_err(|e| SiDbError::LayerDb(e.to_string()))?
+            {
+                return Ok(layer_func_runs.into_iter().map(|arc| (*arc).clone()).collect());
+            }
+        }
+
         Ok(func_runs)
     }
 
@@ -585,6 +644,24 @@ impl FuncRunDb {
             func_runs.push(func_run);
         }
 
+        // Fall back to layer-db if si-db returned no results
+        if func_runs.is_empty() {
+            if let Some(layer_func_runs) = ctx
+                .func_run_layer_db()
+                .read_many_for_component_paginated(
+                    workspace_pk,
+                    change_set_id,
+                    component_id,
+                    limit,
+                    cursor,
+                )
+                .await
+                .map_err(|e| SiDbError::LayerDb(e.to_string()))?
+            {
+                return Ok(layer_func_runs.into_iter().map(|arc| (*arc).clone()).collect());
+            }
+        }
+
         Ok(func_runs)
     }
 }

lib/si-db/src/func_run.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use si_events::{
     FuncRunId,
     FuncRunLog,
@@ -18,11 +20,12 @@ pub struct FuncRunLogDb {}
 
 impl FuncRunLogDb {
     /// Write a new func run log to the database.
-    /// This function can be used to replace the layer-cache write() function.
+    /// This function writes to both si-db and layer-db for backward compatibility during migration.
     pub async fn upsert(ctx: &impl SiDbContext, func_run_log: FuncRunLog) -> SiDbResult<()> {
         let postcard_bytes =
             postcard::to_stdvec(&func_run_log).map_err(|e| SiDbError::Postcard(e.to_string()))?;
 
+        // Write to si-db
         ctx.txns()
             .await?
             .pg()
@@ -63,6 +66,23 @@ impl FuncRunLogDb {
             )
             .await?;
 
+        // Also write to layer-db for backward compatibility during migration
+        // Convert HistoryActor to Actor
+        let actor = match ctx.history_actor() {
+            crate::history_event::HistoryActor::SystemInit => si_events::Actor::System,
+            crate::history_event::HistoryActor::User(pk) => si_events::Actor::User(*pk),
+        };
+
+        ctx.func_run_log_layer_db()
+            .write(
+                Arc::new(func_run_log.clone()),
+                None, // web_events
+                func_run_log.tenancy(),
+                actor,
+            )
+            .await
+            .map_err(|e| SiDbError::LayerDb(e.to_string()))?;
+
         Ok(())
     }
 
@@ -86,7 +106,12 @@ impl FuncRunLogDb {
                 .map_err(|e| SiDbError::Postcard(e.to_string()))?;
             Ok(Some(func_run_log))
         } else {
-            Ok(None)
+            // Fall back to layer-db if not found in si-db
+            ctx.func_run_log_layer_db()
+                .get_for_func_run_id(func_run_id)
+                .await
+                .map(|opt_arc| opt_arc.map(|arc| (*arc).clone()))
+                .map_err(|e| SiDbError::LayerDb(e.to_string()))
         }
     }
 

lib/si-db/src/func_run_log.rs

@@ -61,6 +61,8 @@ mod embedded {
 pub enum SiDbError {
     #[error("action id not found: {0}")]
     ActionIdNotFound(si_events::ActionId),
+    #[error("layer db error: {0}")]
+    LayerDb(String),
     #[error("missing func run: {0}")]
     MissingFuncRun(si_events::FuncRunId),
     #[error("nats error")]