Skip to content

Commit 0e7526d

Browse files
tomasoltomasol
authored
Merge pull request #708 from obeli-sk/deployment-digest
feat(db,api): Add deployment digest
2 parents 6da9bad + 4a23b74 commit 0e7526d

15 files changed

Lines changed: 195 additions & 65 deletions

File tree

‎assets/schemas/openapi.json‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1929,6 +1929,7 @@
19291929
"description": "Deployment details with config",
19301930
"required": [
19311931
"deployment_id",
1932+
"digest",
19321933
"status",
19331934
"created_at",
19341935
"config_json"
@@ -1950,6 +1951,10 @@
19501951
"null"
19511952
]
19521953
},
1954+
"digest": {
1955+
"type": "string",
1956+
"description": "Content digest derived from the deployment's canonical config JSON"
1957+
},
19531958
"last_active_at": {
19541959
"type": [
19551960
"string",
@@ -1967,6 +1972,7 @@
19671972
"description": "Deployment state with execution counts",
19681973
"required": [
19691974
"deployment_id",
1975+
"digest",
19701976
"status",
19711977
"created_at",
19721978
"locked",
@@ -2002,6 +2008,10 @@
20022008
],
20032009
"description": "Optional human-readable deployment description"
20042010
},
2011+
"digest": {
2012+
"type": "string",
2013+
"description": "Content digest derived from the deployment's canonical config JSON"
2014+
},
20052015
"finished_error": {
20062016
"type": "integer",
20072017
"format": "int32",

‎crates/concepts/src/storage.rs‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
use crate::ComponentId;
22
use crate::ComponentRetryConfig;
33
use crate::ComponentType;
4+
use crate::ContentDigest;
45
use crate::ExecutionFailureKind;
56
use crate::ExecutionId;
67
use crate::ExecutionMetadata;
@@ -1569,6 +1570,8 @@ pub const LIST_DEPLOYMENT_STATES_DEFAULT_PAGINATION: Pagination<Option<Deploymen
15691570
pub struct DeploymentState {
15701571
pub deployment_id: DeploymentId,
15711572
pub description: Option<String>,
1573+
/// Content digest derived from the deployment's canonical config JSON.
1574+
pub digest: ContentDigest,
15721575
pub locked: u32,
15731576
// In `PendingAt` state, scheduled to present or past
15741577
pub pending: u32,
@@ -1623,6 +1626,8 @@ impl std::str::FromStr for DeploymentStatus {
16231626
pub struct DeploymentRecord {
16241627
pub deployment_id: DeploymentId,
16251628
pub description: Option<String>,
1629+
/// Content digest derived from `config_json`.
1630+
pub digest: ContentDigest,
16261631
pub created_at: DateTime<Utc>,
16271632
/// Set when the deployment becomes Active; None if it has never been active.
16281633
pub last_active_at: Option<DateTime<Utc>>,
@@ -1632,6 +1637,16 @@ pub struct DeploymentRecord {
16321637
pub created_by: Option<String>,
16331638
}
16341639

1640+
impl DeploymentRecord {
1641+
/// Computes the content digest of a deployment from its canonical config JSON.
1642+
#[must_use]
1643+
pub fn compute_digest(config_json: &str) -> ContentDigest {
1644+
use sha2::{Digest as _, Sha256};
1645+
let hash: [u8; 32] = Sha256::digest(config_json.as_bytes()).into();
1646+
ContentDigest(crate::component_id::Digest(hash))
1647+
}
1648+
}
1649+
16351650
#[derive(Debug, Clone)]
16361651
pub struct ComponentMetadataRecord {
16371652
pub component_digest: ComponentDigest,

‎crates/db-postgres/migrations/V7__deployment_digest.sql‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
ALTER TABLE t_deployment ADD COLUMN digest TEXT NOT NULL DEFAULT 'sha256:0000000000000000000000000000000000000000000000000000000000000000';

‎crates/db-postgres/src/postgres_dao.rs‎

Lines changed: 32 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,9 @@ use crate::postgres_dao::ddl::ADMIN_DB_NAME;
22
use async_trait::async_trait;
33
use chrono::{DateTime, Utc};
44
use concepts::{
5-
ComponentId, ComponentRetryConfig, ComponentType, ExecutionId, FunctionFqn, JoinSetId,
6-
StrVariant, SupportedFunctionReturnValue,
7-
component_id::{ComponentDigest, Digest},
5+
ComponentId, ComponentRetryConfig, ComponentType, ContentDigest, ExecutionId, FunctionFqn,
6+
JoinSetId, StrVariant, SupportedFunctionReturnValue,
7+
component_id::ComponentDigest,
88
prefixed_ulid::{DelayId, DeploymentId, ExecutionIdDerived, ExecutorId, RunId},
99
storage::{
1010
AppendBatchResponse, AppendDelayResponseOutcome, AppendEventsToExecution, AppendRequest,
@@ -297,6 +297,10 @@ impl PostgresPool {
297297
}
298298
}
299299

300+
fn deployment_digest_from_pg_row(row: &Row) -> Result<ContentDigest, DbErrorRead> {
301+
Ok(get(row, "digest")?)
302+
}
303+
300304
fn deployment_record_from_pg_row(row: &Row) -> Result<DeploymentRecord, DbErrorRead> {
301305
let deployment_id_str: String = get(row, "deployment_id")?;
302306
let deployment_id = deployment_id_str.parse::<DeploymentId>().map_err(|e| {
@@ -309,6 +313,7 @@ fn deployment_record_from_pg_row(row: &Row) -> Result<DeploymentRecord, DbErrorR
309313
Ok(DeploymentRecord {
310314
deployment_id,
311315
description: get(row, "description")?,
316+
digest: get(row, "digest")?,
312317
created_at: get(row, "created_at")?,
313318
last_active_at: get(row, "last_active_at")?,
314319
status,
@@ -322,19 +327,8 @@ fn deployment_component_detail_from_pg_row(
322327
row: &Row,
323328
) -> Result<DeploymentComponentDetail, DbErrorRead> {
324329
let component_name: String = get(row, "component_name")?;
325-
let component_type: ComponentType =
326-
get::<String, _>(row, "component_type")?
327-
.parse()
328-
.map_err(|err| {
329-
DbErrorRead::Generic(consistency_db_err(format!("invalid component_type: {err}")))
330-
})?;
331-
let component_digest = ComponentDigest(Digest(
332-
get::<Vec<u8>, _>(row, "component_digest")?
333-
.try_into()
334-
.map_err(|_| {
335-
DbErrorRead::Generic(consistency_db_err("invalid component_digest length"))
336-
})?,
337-
));
330+
let component_type: ComponentType = get(row, "component_type")?;
331+
let component_digest: ComponentDigest = get(row, "component_digest")?;
338332
let component_id = ComponentId::new(
339333
component_type,
340334
StrVariant::from(component_name),
@@ -1062,24 +1056,15 @@ async fn get_combined_state(
10621056
let created_at: DateTime<Utc> = get(&row, "created_at")?;
10631057
let first_scheduled_at: DateTime<Utc> = get(&row, "first_scheduled_at")?;
10641058

1065-
let digest_bytes: Vec<u8> = get(&row, "component_id_input_digest")?;
1066-
let digest = Digest::try_from(digest_bytes.as_slice()).map_err(|err| {
1067-
consistency_db_err_src("cannot parse `component_id_input_digest`", Arc::from(err))
1068-
})?;
1069-
let component_digest = ComponentDigest(digest);
1059+
let component_digest: ComponentDigest = get(&row, "component_id_input_digest")?;
10701060

1071-
let component_type: String = get(&row, "component_type")?;
1072-
let component_type = ComponentType::from_str(&component_type)
1073-
.map_err(|err| consistency_db_err_src("cannot parse `component_type`", Arc::from(err)))?;
1061+
let component_type: ComponentType = get(&row, "component_type")?;
10741062

10751063
let deployment_id: String = get(&row, "deployment_id")?;
10761064
let deployment_id = DeploymentId::from_str(&deployment_id).map_err(DbErrorGeneric::from)?;
10771065

10781066
let state: String = get(&row, "state")?;
1079-
let ffqn: String = get(&row, "ffqn")?;
1080-
let ffqn = FunctionFqn::from_str(&ffqn).map_err(|parse_err| {
1081-
consistency_db_err(format!("invalid ffqn value in `t_state` - {parse_err}"))
1082-
})?;
1067+
let ffqn: FunctionFqn = get(&row, "ffqn")?;
10831068

10841069
let pending_expires_finished: DateTime<Utc> = get(&row, "pending_expires_finished")?;
10851070

@@ -1283,16 +1268,9 @@ async fn list_executions(
12831268
let execution_id = ExecutionId::from_str(&execution_id_str)
12841269
.map_err(|err| consistency_db_err(err.to_string()))?;
12851270

1286-
let digest_bytes: Vec<u8> = get(&row, "component_id_input_digest")?;
1287-
let digest = Digest::try_from(digest_bytes.as_slice()).map_err(|err| {
1288-
consistency_db_err_src("cannot parse `component_id_input_digest`", Arc::from(err))
1289-
})?;
1290-
let component_digest = ComponentDigest(digest);
1271+
let component_digest: ComponentDigest = get(&row, "component_id_input_digest")?;
12911272

1292-
let component_type: String = get(&row, "component_type")?;
1293-
let component_type = ComponentType::from_str(&component_type).map_err(|err| {
1294-
consistency_db_err_src("cannot parse `component_type`", Arc::from(err))
1295-
})?;
1273+
let component_type: ComponentType = get(&row, "component_type")?;
12961274

12971275
let deployment_id: String = get(&row, "deployment_id")?;
12981276
let deployment_id =
@@ -1330,11 +1308,7 @@ async fn list_executions(
13301308
.map(|id| JoinSetId::from_str(&id))
13311309
.transpose()?;
13321310

1333-
let ffqn: String = get(&row, "ffqn")?;
1334-
let ffqn = FunctionFqn::from_str(&ffqn).map_err(|parse_err| {
1335-
error!("Error parsing ffqn - {parse_err:?}");
1336-
consistency_db_err("invalid ffqn value in `t_state`")
1337-
})?;
1311+
let ffqn: FunctionFqn = get(&row, "ffqn")?;
13381312

13391313
let combined_state_dto = CombinedStateDTO {
13401314
execution_id,
@@ -1673,6 +1647,7 @@ async fn list_deployment_states(
16731647
SELECT
16741648
d.deployment_id,
16751649
d.description,
1650+
d.digest,
16761651
16771652
COUNT(*) FILTER (WHERE s.state = '{STATE_LOCKED}' AND s.is_paused = false) AS locked,
16781653
@@ -1743,7 +1718,7 @@ async fn list_deployment_states(
17431718

17441719
write!(
17451720
sql,
1746-
" GROUP BY d.deployment_id, d.description, d.config_json, d.created_at, d.last_active_at, d.status ORDER BY d.deployment_id {inner_order} LIMIT {}",
1721+
" GROUP BY d.deployment_id, d.description, d.digest, d.config_json, d.created_at, d.last_active_at, d.status ORDER BY d.deployment_id {inner_order} LIMIT {}",
17471722
pagination.length()
17481723
)
17491724
.expect("writing to string");
@@ -1774,6 +1749,7 @@ async fn list_deployment_states(
17741749
result.push(DeploymentState {
17751750
deployment_id: DeploymentId::from_str(&deployment_id).map_err(DbErrorGeneric::from)?,
17761751
description: get::<Option<String>, _>(&row, "description")?,
1752+
digest: deployment_digest_from_pg_row(&row)?,
17771753
locked: u32::try_from(get::<i64, _>(&row, "locked")?).expect("count is never negative"),
17781754
pending: u32::try_from(get::<i64, _>(&row, "pending")?)
17791755
.expect("count is never negative"),
@@ -4681,18 +4657,20 @@ impl DbExternalApi for PostgresConnection {
46814657
);
46824658
let mut client_guard = self.client.lock().await;
46834659
let tx = client_guard.transaction().await?;
4660+
let digest = record.digest.to_string();
46844661
tx.execute(
46854662
"INSERT INTO t_deployment \
4686-
(deployment_id, description, created_at, status, config_json, obelisk_version, created_by) \
4687-
VALUES ($1, $2, $3, $4, $5, $6, $7)",
4663+
(deployment_id, description, digest, created_at, status, config_json, obelisk_version, created_by) \
4664+
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
46884665
&[
46894666
&record.deployment_id.to_string(), // $1
46904667
&record.description, // $2
4691-
&record.created_at, // $3
4692-
&record.status.as_str(), // $4
4693-
&record.config_json, // $5
4694-
&record.obelisk_version, // $6
4695-
&record.created_by, // $7
4668+
&digest, // $3
4669+
&record.created_at, // $4
4670+
&record.status.as_str(), // $5
4671+
&record.config_json, // $6
4672+
&record.obelisk_version, // $7
4673+
&record.created_by, // $8
46964674
],
46974675
)
46984676
.await?;
@@ -4772,7 +4750,7 @@ impl DbExternalApi for PostgresConnection {
47724750
let tx = client_guard.transaction().await?;
47734751
let row = tx
47744752
.query_opt(
4775-
"SELECT deployment_id, description, created_at, last_active_at, status, config_json, obelisk_version, created_by \
4753+
"SELECT deployment_id, description, digest, created_at, last_active_at, status, config_json, obelisk_version, created_by \
47764754
FROM t_deployment WHERE deployment_id = $1",
47774755
&[&deployment_id.to_string()],
47784756
)
@@ -4791,7 +4769,7 @@ impl DbExternalApi for PostgresConnection {
47914769
let tx = client_guard.transaction().await?;
47924770
let row = tx
47934771
.query_opt(
4794-
"SELECT deployment_id, description, created_at, last_active_at, status, config_json, obelisk_version, created_by \
4772+
"SELECT deployment_id, description, digest, created_at, last_active_at, status, config_json, obelisk_version, created_by \
47954773
FROM t_deployment WHERE status = 'active' LIMIT 1",
47964774
&[],
47974775
)
@@ -4808,7 +4786,7 @@ impl DbExternalApi for PostgresConnection {
48084786
let tx = client_guard.transaction().await?;
48094787
let row = tx
48104788
.query_opt(
4811-
"SELECT deployment_id, description, created_at, last_active_at, status, config_json, obelisk_version, created_by \
4789+
"SELECT deployment_id, description, digest, created_at, last_active_at, status, config_json, obelisk_version, created_by \
48124790
FROM t_deployment WHERE status IN ('enqueued', 'active') \
48134791
ORDER BY CASE status WHEN 'enqueued' THEN 0 ELSE 1 END LIMIT 1",
48144792
&[],
@@ -4835,7 +4813,7 @@ impl DbExternalApi for PostgresConnection {
48354813
};
48364814

48374815
let mut sql = String::from(
4838-
"SELECT deployment_id, description, created_at, last_active_at, status, config_json, obelisk_version, created_by \
4816+
"SELECT deployment_id, description, digest, created_at, last_active_at, status, config_json, obelisk_version, created_by \
48394817
FROM t_deployment",
48404818
);
48414819

‎crates/db-sqlite/migrations/V7__deployment_digest.sql‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
ALTER TABLE t_deployment ADD COLUMN digest TEXT NOT NULL DEFAULT 'sha256:0000000000000000000000000000000000000000000000000000000000000000';

‎crates/db-sqlite/src/sqlite_dao.rs‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -425,6 +425,7 @@ fn deployment_record_from_row(row: &Row<'_>) -> rusqlite::Result<DeploymentRecor
425425
Ok(DeploymentRecord {
426426
deployment_id,
427427
description: row.get("description")?,
428+
digest: row.get("digest")?,
428429
created_at: row.get("created_at")?,
429430
last_active_at: row.get("last_active_at")?,
430431
status,
@@ -3316,6 +3317,7 @@ impl SqlitePool {
33163317
SELECT
33173318
d.deployment_id,
33183319
d.description,
3320+
d.digest,
33193321
COALESCE(SUM(s.state = '{STATE_LOCKED}' AND s.is_paused = false), 0) AS locked,
33203322
COALESCE(SUM(s.state = '{STATE_PENDING_AT}' AND s.is_paused = false AND s.pending_expires_finished <= :now), 0) AS pending,
33213323
COALESCE(SUM(s.state = '{STATE_PENDING_AT}' AND s.is_paused = false AND s.pending_expires_finished > :now), 0) AS scheduled,
@@ -3360,7 +3362,7 @@ impl SqlitePool {
33603362

33613363
write!(
33623364
sql,
3363-
" GROUP BY d.deployment_id, d.description, d.config_json, d.created_at, d.last_active_at, d.status ORDER BY d.deployment_id {inner_order} LIMIT {limit}",
3365+
" GROUP BY d.deployment_id, d.description, d.digest, d.config_json, d.created_at, d.last_active_at, d.status ORDER BY d.deployment_id {inner_order} LIMIT {limit}",
33643366
limit = pagination.length()
33653367
)
33663368
.expect("writing to string");
@@ -3391,6 +3393,7 @@ impl SqlitePool {
33913393
Ok(DeploymentState {
33923394
deployment_id: row.get("deployment_id")?,
33933395
description: row.get("description")?,
3396+
digest: row.get("digest")?,
33943397
locked: row.get("locked")?,
33953398
pending: row.get("pending")?,
33963399
scheduled: row.get("scheduled")?,
@@ -3427,11 +3430,12 @@ impl SqlitePool {
34273430
);
34283431
tx.execute(
34293432
"INSERT INTO t_deployment \
3430-
(deployment_id, description, created_at, status, config_json, obelisk_version, created_by) \
3431-
VALUES (:deployment_id, :description, :created_at, :status, :config_json, :obelisk_version, :created_by)",
3433+
(deployment_id, description, digest, created_at, status, config_json, obelisk_version, created_by) \
3434+
VALUES (:deployment_id, :description, :digest, :created_at, :status, :config_json, :obelisk_version, :created_by)",
34323435
rusqlite::named_params! {
34333436
":deployment_id": record.deployment_id.to_string(),
34343437
":description": record.description,
3438+
":digest": record.digest.to_string(),
34353439
":created_at": record.created_at,
34363440
":status": record.status.as_str(),
34373441
":config_json": record.config_json,
@@ -3514,7 +3518,7 @@ impl SqlitePool {
35143518
deployment_id: DeploymentId,
35153519
) -> Result<Option<DeploymentRecord>, DbErrorRead> {
35163520
tx.query_row(
3517-
"SELECT deployment_id, description, created_at, last_active_at, status, config_json, obelisk_version, created_by \
3521+
"SELECT deployment_id, description, digest, created_at, last_active_at, status, config_json, obelisk_version, created_by \
35183522
FROM t_deployment WHERE deployment_id = :deployment_id",
35193523
rusqlite::named_params! { ":deployment_id": deployment_id.to_string() },
35203524
deployment_record_from_row,
@@ -3526,7 +3530,7 @@ impl SqlitePool {
35263530
#[cfg(feature = "test")]
35273531
fn get_active_deployment_tx(tx: &Transaction) -> Result<Option<DeploymentRecord>, DbErrorRead> {
35283532
tx.query_row(
3529-
"SELECT deployment_id, description, created_at, last_active_at, status, config_json, obelisk_version, created_by \
3533+
"SELECT deployment_id, description, digest, created_at, last_active_at, status, config_json, obelisk_version, created_by \
35303534
FROM t_deployment WHERE status = 'active' LIMIT 1",
35313535
[],
35323536
deployment_record_from_row,
@@ -3541,7 +3545,7 @@ impl SqlitePool {
35413545
) -> Result<Vec<DeploymentRecord>, DbErrorRead> {
35423546
let mut params: Vec<(&'static str, Box<dyn ToSql>)> = vec![];
35433547
let mut sql = String::from(
3544-
"SELECT deployment_id, description, created_at, last_active_at, status, config_json, obelisk_version, created_by \
3548+
"SELECT deployment_id, description, digest, created_at, last_active_at, status, config_json, obelisk_version, created_by \
35453549
FROM t_deployment",
35463550
);
35473551

@@ -4634,7 +4638,7 @@ impl DbExternalApi for SqlitePool {
46344638
self.transaction(
46354639
move |tx| {
46364640
tx.query_row(
4637-
"SELECT deployment_id, description, created_at, last_active_at, status, config_json, obelisk_version, created_by \
4641+
"SELECT deployment_id, description, digest, created_at, last_active_at, status, config_json, obelisk_version, created_by \
46384642
FROM t_deployment WHERE status IN ('enqueued', 'active') \
46394643
ORDER BY CASE status WHEN 'enqueued' THEN 0 ELSE 1 END LIMIT 1",
46404644
[],

0 commit comments

Comments
 (0)