sayiir_postgres/
task_result_store.rs

1//! [`TaskResultStore`] implementation for Postgres.
2//!
3//! For non-terminal workflows, task results come from the current snapshot.
4//! For completed or failed workflows (where `completed_tasks` has been
5//! discarded), the implementation falls back to the most recent `InProgress`
6//! entry in `sayiir_workflow_snapshot_history`.
7
8use sayiir_core::codec::{self, Decoder};
9use sayiir_core::snapshot::WorkflowSnapshot;
10use sayiir_persistence::{BackendError, SnapshotStore, TaskResultStore};
11use sqlx::Row;
12
13use crate::backend::PostgresBackend;
14use crate::error::PgError;
15
16impl<C> TaskResultStore for PostgresBackend<C>
17where
18    C: codec::SnapshotCodec,
19{
20    #[tracing::instrument(
21        name = "db.load_task_result",
22        skip(self),
23        fields(db.system = "postgresql"),
24        err(level = tracing::Level::ERROR),
25    )]
26    async fn load_task_result(
27        &self,
28        instance_id: &str,
29        task_id: &sayiir_core::TaskId,
30    ) -> Result<Option<bytes::Bytes>, BackendError> {
31        let snapshot = self.load_snapshot(instance_id).await?;
32
33        // Non-terminal states carry completed_tasks directly.
34        if let Some(bytes) = snapshot.get_task_result_bytes(task_id) {
35            return Ok(Some(bytes));
36        }
37
38        // For terminal states, fall back to snapshot history.
39        if (snapshot.state.is_completed() || snapshot.state.is_failed())
40            && let Some(hist) = self.load_last_in_progress_snapshot(instance_id).await?
41        {
42            return Ok(hist.get_task_result_bytes(task_id));
43        }
44
45        Ok(None)
46    }
47}
48
49impl<C> PostgresBackend<C>
50where
51    C: Decoder + codec::CodecIdentity + codec::sealed::DecodeValue<WorkflowSnapshot>,
52{
53    /// Load the most recent `InProgress` snapshot from the history table.
54    ///
55    /// This is the last snapshot before the workflow transitioned to a terminal
56    /// state, so it still contains `completed_tasks` — but the on-disk blob
57    /// is outputs-stripped, so hydrate per-task outputs from
58    /// `sayiir_workflow_tasks` before returning.
59    async fn load_last_in_progress_snapshot(
60        &self,
61        instance_id: &str,
62    ) -> Result<Option<WorkflowSnapshot>, BackendError> {
63        let row = sqlx::query(
64            "SELECT data FROM sayiir_workflow_snapshot_history
65             WHERE instance_id = $1 AND status = 'InProgress'
66             ORDER BY version DESC LIMIT 1",
67        )
68        .bind(instance_id)
69        .fetch_optional(&self.pool)
70        .await
71        .map_err(PgError)?;
72
73        match row {
74            Some(r) => {
75                let raw: &[u8] = r.get("data");
76                let mut snapshot = self.decode(raw)?;
77                let outputs = crate::history::fetch_task_outputs(&self.pool, instance_id).await?;
78                snapshot.hydrate_task_outputs(outputs);
79                Ok(Some(snapshot))
80            }
81            None => Ok(None),
82        }
83    }
84}
sayiir_postgres/task_result_store.rs

sayiir_postgres/
task_result_store.rs
