Merge pull request #359 from obeli-sk/persist-sources

feat(sqlite,pg): Store sources and frame source mappings in db

Author: tomasol

Cargo.lock

@@ -43,7 +43,10 @@ Or manually install dependencies (see [dev-deps.txt](dev-deps.txt)).
 Postgres must be running. See `.envrc-example` for environment variables.
 ```sh
 # All tests
-./scripts/test.sh
+scripts/test.sh
+
+# Tests with locally built activity, workflow and webhook JavaScript runtimes
+scripts/test-js-local.sh
 
 # Specific test crate
 cargo test --package obeli-db-tests --test deployment_pagination
@@ -52,35 +55,17 @@ cargo test --package obeli-db-tests --test deployment_pagination
 cargo test --package obelisk grpc_server::tests
 ```
 
-
-## Code Patterns
-
-### WASM Component Error Variants
-
-WASM components returning `result<T, E>` where `E` is a variant type must include an
-`execution-failed` case with no payload. `E` can also be a `string` or an empty type
-(see `enum ReturnType`).
-
-### Adding Workflow Host Functions
-
-WIT definitions:
-- Types: `wit/obelisk_types@X.Y.Z/types.wit` (symlinked as `@latest`)
-- Workflow support: `wit/obelisk_workflow@X.Y.Z/workflow-support.wit` (symlinked as `@latest`)
-
-Implementation pattern:
-1. Define function signature in WIT with `@since` annotation
-2. Implement in `crates/wasm-workers/src/workflow/workflow_ctx.rs`
-3. Link via `add_to_linker_workflow_support()` using `func_wrap` or `func_wrap_async`
-4. Type conversions in `host_exports.rs` (WIT ↔ Rust types via wasmtime bindgen)
-5. Test with programs in `crates/testing/test-programs/`
-
 ## Key Files
 
 | Task | Files |
 |------|-------|
+| TOML | `src/config/toml.rs`, `obelisk-help.toml` |
 | Database schema/queries | `crates/db-sqlite/src/sqlite_dao.rs`, `crates/db-postgres/src/postgres_dao.rs` |
 | Storage traits | `crates/concepts/src/storage.rs` |
-| gRPC API | `proto/obelisk.proto`, `src/server/grpc_server.rs` |
+| gRPC API | `proto/obelisk.proto`, `src/server/grpc_server.rs`,`crates/grpc/src/grpc_mapping.rs` |
 | REST API | `src/server/web_api_server.rs` |
-| Type conversions (gRPC) | `crates/grpc/src/grpc_mapping.rs` |
+| Server | `src/command/server.rs`, `crates/wasm-workers/src/registry.rs` |
+| Activities | `crates/wasm-workers/src/activity/` |
+| Workflows | `crates/wasm-workers/src/workflow/` |
+| Webhooks | `crates/wasm-workers/src/webhook/` |
 | Test utilities | `crates/testing/test-utils/src/` |

DEVELOPMENT.md

@@ -1176,6 +1176,26 @@ pub trait DbExternalApi: DbConnection {
         filter: BacktraceFilter,
     ) -> Result<BacktraceInfo, DbErrorRead>;
 
+    /// Store a source file associated with a component digest.
+    /// `frame_key` is either an exact frame symbol path or a suffix (with leading `/`)
+    /// when `is_suffix` is true. Idempotent — repeated calls for the same key are ignored.
+    async fn upsert_source_file(
+        &self,
+        component_digest: &ComponentDigest,
+        frame_key: &str,
+        is_suffix: bool,
+        content: &str,
+    ) -> Result<(), DbErrorWrite>;
+
+    /// Look up a source file by component digest and a frame symbol path.
+    /// Matches either exact keys or suffix keys (where the frame path ends with the stored key).
+    /// Returns `None` if not found or if multiple suffix entries match (ambiguous).
+    async fn get_source_file(
+        &self,
+        component_digest: &ComponentDigest,
+        file: &str,
+    ) -> Result<Option<String>, DbErrorRead>;
+
     /// Returns executions sorted in descending order.
     async fn list_executions(
         &self,

crates/concepts/src/storage.rs

@@ -26,6 +26,7 @@ secrecy.workspace = true
 strum.workspace = true
 tokio-postgres.workspace = true
 rand.workspace = true
+sha2.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
 tracing.workspace = true

crates/db-postgres/Cargo.toml

@@ -30,6 +30,7 @@ use db_common::{
 use deadpool_postgres::{Client, ManagerConfig, Pool, RecyclingMethod};
 use hashbrown::HashMap;
 use secrecy::{ExposeSecret as _, SecretString};
+use sha2::{Digest as _, Sha256};
 use std::{collections::VecDeque, pin::Pin, str::FromStr as _, sync::Arc, time::Duration};
 use std::{fmt::Write as _, panic::Location};
 use strum::IntoEnumIterator as _;
@@ -239,6 +240,24 @@ CREATE TABLE IF NOT EXISTS t_execution_backtrace (
 
     pub const IDX_T_EXECUTION_BACKTRACE_EXECUTION_ID_VERSION: &str = r"
 CREATE INDEX IF NOT EXISTS idx_t_execution_backtrace_execution_id_version ON t_execution_backtrace (execution_id, version_min_including, version_max_excluding);
+";
+
+    // Source files
+    pub const CREATE_TABLE_T_SOURCE_FILE: &str = r"
+CREATE TABLE IF NOT EXISTS t_source_file (
+    content_hash BYTEA PRIMARY KEY,
+    content      TEXT NOT NULL
+);
+";
+    pub const CREATE_TABLE_T_COMPONENT_SOURCE: &str = r"
+CREATE TABLE IF NOT EXISTS t_component_source (
+    component_digest BYTEA   NOT NULL,
+    frame_key        TEXT    NOT NULL,
+    is_suffix        BOOLEAN NOT NULL,
+    content_hash     BYTEA   NOT NULL,
+    PRIMARY KEY (component_digest, frame_key, is_suffix),
+    FOREIGN KEY (content_hash) REFERENCES t_source_file(content_hash)
+);
 ";
 
     // Logs & Std sterams
@@ -487,6 +506,8 @@ impl PostgresPool {
             ddl::CREATE_TABLE_T_WASM_BACKTRACE,
             ddl::CREATE_TABLE_T_EXECUTION_BACKTRACE,
             ddl::IDX_T_EXECUTION_BACKTRACE_EXECUTION_ID_VERSION,
+            ddl::CREATE_TABLE_T_SOURCE_FILE,
+            ddl::CREATE_TABLE_T_COMPONENT_SOURCE,
             ddl::CREATE_TABLE_T_LOG,
             ddl::IDX_T_LOG_EXECUTION_ID_RUN_ID_CREATED_AT,
             ddl::IDX_T_LOG_EXECUTION_ID_CREATED_AT,
@@ -4062,6 +4083,73 @@ impl DbExternalApi for PostgresConnection {
         })
     }
 
+    #[instrument(skip_all)]
+    async fn upsert_source_file(
+        &self,
+        component_digest: &ComponentDigest,
+        frame_key: &str,
+        is_suffix: bool,
+        content: &str,
+    ) -> Result<(), DbErrorWrite> {
+        let content_hash: [u8; 32] = Sha256::digest(content.as_bytes()).into();
+        let mut client_guard = self.client.lock().await;
+        let tx = client_guard.transaction().await?;
+        tx.execute(
+            "INSERT INTO t_source_file (content_hash, content) \
+             VALUES ($1, $2) \
+             ON CONFLICT (content_hash) DO NOTHING",
+            &[&content_hash.as_slice(), &content],
+        )
+        .await?;
+        tx.execute(
+            "INSERT INTO t_component_source \
+             (component_digest, frame_key, is_suffix, content_hash) \
+             VALUES ($1, $2, $3, $4) \
+             ON CONFLICT (component_digest, frame_key, is_suffix) DO NOTHING",
+            &[
+                &component_digest.as_slice(),
+                &frame_key,
+                &is_suffix,
+                &content_hash.as_slice(),
+            ],
+        )
+        .await?;
+        tx.commit().await?;
+        Ok(())
+    }
+
+    #[instrument(skip_all)]
+    async fn get_source_file(
+        &self,
+        component_digest: &ComponentDigest,
+        file: &str,
+    ) -> Result<Option<String>, DbErrorRead> {
+        let mut client_guard = self.client.lock().await;
+        let tx = client_guard.transaction().await?;
+        let rows = tx
+            .query(
+                "SELECT s.content \
+                 FROM t_component_source cs \
+                 JOIN t_source_file s ON cs.content_hash = s.content_hash \
+                 WHERE cs.component_digest = $1 \
+                   AND ( \
+                       (NOT cs.is_suffix AND cs.frame_key = $2) \
+                    OR (cs.is_suffix AND right($2, length(cs.frame_key)) = cs.frame_key) \
+                   )",
+                &[&component_digest.as_slice(), &file],
+            )
+            .await?;
+        tx.commit().await?;
+        match rows.len() {
+            0 => Ok(None),
+            1 => Ok(Some(get::<String, _>(&rows[0], "content")?)),
+            _ => {
+                warn!("Multiple suffix matches for '{file}', returning None");
+                Ok(None)
+            }
+        }
+    }
+
     #[instrument(skip(self))]
     async fn list_executions(
         &self,

crates/db-postgres/src/postgres_dao.rs

@@ -25,6 +25,7 @@ itertools.workspace = true
 rand.workspace = true
 rusqlite.workspace = true
 serde_json.workspace = true
+sha2.workspace = true
 serde.workspace = true
 strum.workspace = true
 tempfile = { workspace = true, optional = true }

crates/db-sqlite/Cargo.toml

@@ -34,6 +34,7 @@ use rusqlite::{
     CachedStatement, Connection, OpenFlags, OptionalExtension, Params, Row, ToSql, Transaction,
     TransactionBehavior, named_params, types::ToSqlOutput,
 };
+use sha2::{Digest as _, Sha256};
 use std::{
     cmp::max,
     collections::VecDeque,
@@ -303,6 +304,30 @@ CREATE TABLE IF NOT EXISTS t_wasm_backtrace (
 ) STRICT
 ";
 
+// Content-addressed store for source file text. Content hash is SHA-256 of the UTF-8 content.
+const CREATE_TABLE_T_SOURCE_FILE: &str = r"
+CREATE TABLE IF NOT EXISTS t_source_file (
+    content_hash BLOB NOT NULL,
+    content      TEXT NOT NULL,
+
+    PRIMARY KEY (content_hash)
+) STRICT
+";
+// Maps (component_digest, frame_key, is_suffix) to a source file.
+// frame_key is the exact frame symbol path (is_suffix=0) or a '/'-prefixed suffix (is_suffix=1).
+const CREATE_TABLE_T_COMPONENT_SOURCE: &str = r"
+CREATE TABLE IF NOT EXISTS t_component_source (
+    component_digest BLOB    NOT NULL,
+    frame_key        TEXT    NOT NULL,
+    is_suffix        INTEGER NOT NULL,
+    content_hash     BLOB    NOT NULL,
+
+    PRIMARY KEY (component_digest, frame_key, is_suffix),
+    FOREIGN KEY (content_hash)
+        REFERENCES t_source_file(content_hash)
+) STRICT
+";
+
 /// Stores logs and std stream output of execution runs. Append only.
 /// Logs have `level` and `message` null.
 /// Std streams have `stream_type`, `payload` not null.
@@ -779,6 +804,9 @@ impl SqlitePool {
         conn_execute(&conn, CREATE_TABLE_T_EXECUTION_BACKTRACE, [])?;
         conn_execute(&conn, IDX_T_EXECUTION_BACKTRACE_EXECUTION_ID_VERSION, [])?;
         conn_execute(&conn, CREATE_TABLE_T_WASM_BACKTRACE, [])?;
+        // source files
+        conn_execute(&conn, CREATE_TABLE_T_SOURCE_FILE, [])?;
+        conn_execute(&conn, CREATE_TABLE_T_COMPONENT_SOURCE, [])?;
         // t_log
         conn_execute(&conn, CREATE_TABLE_T_LOG, [])?;
         conn_execute(&conn, IDX_T_LOG_EXECUTION_ID_RUN_ID_CREATED_AT, [])?;
@@ -3862,6 +3890,92 @@ impl DbExternalApi for SqlitePool {
         ).await
     }
 
+    #[instrument(skip_all)]
+    async fn upsert_source_file(
+        &self,
+        component_digest: &ComponentDigest,
+        frame_key: &str,
+        is_suffix: bool,
+        content: &str,
+    ) -> Result<(), DbErrorWrite> {
+        let content_hash: [u8; 32] = Sha256::digest(content.as_bytes()).into();
+        let component_digest = component_digest.clone();
+        let frame_key = frame_key.to_owned();
+        let content = content.to_owned();
+        self.transaction(
+            move |tx| {
+                tx.prepare(
+                    "INSERT OR IGNORE INTO t_source_file (content_hash, content) \
+                     VALUES (:content_hash, :content)",
+                )?
+                .execute(named_params! {
+                    ":content_hash": content_hash,
+                    ":content": content,
+                })?;
+                tx.prepare(
+                    "INSERT OR IGNORE INTO t_component_source \
+                     (component_digest, frame_key, is_suffix, content_hash) \
+                     VALUES (:component_digest, :frame_key, :is_suffix, :content_hash)",
+                )?
+                .execute(named_params! {
+                    ":component_digest": component_digest,
+                    ":frame_key": frame_key,
+                    ":is_suffix": is_suffix,
+                    ":content_hash": content_hash,
+                })?;
+                Ok(())
+            },
+            TxType::Other,
+            "upsert_source_file",
+        )
+        .await
+    }
+
+    #[instrument(skip_all)]
+    async fn get_source_file(
+        &self,
+        component_digest: &ComponentDigest,
+        file: &str,
+    ) -> Result<Option<String>, DbErrorRead> {
+        let component_digest = component_digest.clone();
+        let file = file.to_owned();
+        self.transaction(
+            move |tx| {
+                let mut stmt = tx.prepare(
+                    "SELECT s.content \
+                     FROM t_component_source cs \
+                     JOIN t_source_file s ON cs.content_hash = s.content_hash \
+                     WHERE cs.component_digest = :component_digest \
+                       AND ( \
+                           (cs.is_suffix = 0 AND cs.frame_key = :file) \
+                        OR (cs.is_suffix = 1 AND \
+                            substr(:file, length(:file) - length(cs.frame_key) + 1) = cs.frame_key) \
+                       )",
+                )?;
+                let rows: Vec<String> = stmt
+                    .query_map(
+                        named_params! {
+                            ":component_digest": component_digest,
+                            ":file": file,
+                        },
+                        |row| row.get(0),
+                    )?
+                    .collect::<Result<_, _>>()?;
+                match rows.len() {
+                    0 => Ok(None),
+                    1 => Ok(Some(rows.into_iter().next().unwrap())),
+                    _ => {
+                        warn!("Multiple suffix matches for '{file}', returning None");
+                        Ok(None)
+                    }
+                }
+            },
+            TxType::Other,
+            "get_source_file",
+        )
+        .await
+    }
+
     #[instrument(skip(self))]
     async fn list_executions(
         &self,

crates/db-sqlite/src/sqlite_dao.rs

@@ -35,7 +35,6 @@ pub fn compile_activity_stub_inline(
         workflow_or_activity_config: Some(component_config_importable),
         wit: wit_text_with_extensions,
         workflow_replay_info: None,
-        source: None,
     })
 }