bsodfather/crank
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

core: type streaming session timestamps

Browse Source
This commit is contained in:
a.tolmachev
2026-04-13 05:47:42 +00:00
parent edf318ba67
commit 03cb109b40
10 changed files with 254 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/crank-core/src/stream_session.rs
+50 -36
View File
@@ -1,5 +1,6 @@
use serde::{Deserialize, Serialize};
use serde_json::Value;
use time::OffsetDateTime;
use crate::{
ids::{AgentId, AsyncJobId, OperationId, StreamSessionId, WorkspaceId},
@@ -36,31 +37,34 @@ pub struct StreamSession {
#[serde(skip_serializing_if = "Option::is_none")]
pub cursor: Option<Value>,
pub state: Value,
pub expires_at: String,
#[serde(with = "time::serde::rfc3339")]
pub expires_at: OffsetDateTime,
#[serde(skip_serializing_if = "Option::is_none")]
pub last_poll_at: Option<String>,
pub created_at: String,
#[serde(with = "time::serde::rfc3339::option")]
pub last_poll_at: Option<OffsetDateTime>,
#[serde(with = "time::serde::rfc3339")]
pub created_at: OffsetDateTime,
#[serde(skip_serializing_if = "Option::is_none")]
pub closed_at: Option<String>,
#[serde(with = "time::serde::rfc3339::option")]
pub closed_at: Option<OffsetDateTime>,
}
impl StreamSession {
pub fn is_expired(&self, now: &str) -> bool {
self.expires_at.as_str() <= now
pub fn is_expired(&self, now: OffsetDateTime) -> bool {
self.expires_at <= now
}
pub fn can_poll(&self, now: &str) -> bool {
pub fn can_poll(&self, now: OffsetDateTime) -> bool {
!self.status.is_terminal() && !self.is_expired(now)
}
pub fn mark_polled(&mut self, now: impl Into<String>) {
self.last_poll_at = Some(now.into());
pub fn mark_polled(&mut self, now: OffsetDateTime) {
self.last_poll_at = Some(now);
}
pub fn mark_closed(&mut self, now: impl Into<String>) {
let now = now.into();
pub fn mark_closed(&mut self, now: OffsetDateTime) {
self.status = StreamStatus::Stopped;
self.last_poll_at = Some(now.clone());
self.last_poll_at = Some(now);
self.closed_at = Some(now);
}
}
@@ -99,11 +103,15 @@ pub struct AsyncJobHandle {
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub expires_at: Option<String>,
pub created_at: String,
pub updated_at: String,
#[serde(with = "time::serde::rfc3339::option")]
pub expires_at: Option<OffsetDateTime>,
#[serde(with = "time::serde::rfc3339")]
pub created_at: OffsetDateTime,
#[serde(with = "time::serde::rfc3339")]
pub updated_at: OffsetDateTime,
#[serde(skip_serializing_if = "Option::is_none")]
pub finished_at: Option<String>,
#[serde(with = "time::serde::rfc3339::option")]
pub finished_at: Option<OffsetDateTime>,
}
impl AsyncJobHandle {
@@ -115,20 +123,18 @@ impl AsyncJobHandle {
matches!(self.status, JobStatus::Created | JobStatus::Running)
}
pub fn mark_finished(&mut self, now: impl Into<String>, result: Value) {
let now = now.into();
pub fn mark_finished(&mut self, now: OffsetDateTime, result: Value) {
self.status = JobStatus::Completed;
self.result = Some(result);
self.error = None;
self.updated_at = now.clone();
self.updated_at = now;
self.finished_at = Some(now);
}
pub fn mark_failed(&mut self, now: impl Into<String>, error: Value) {
let now = now.into();
pub fn mark_failed(&mut self, now: OffsetDateTime, error: Value) {
self.status = JobStatus::Failed;
self.error = Some(error);
self.updated_at = now.clone();
self.updated_at = now;
self.finished_at = Some(now);
}
}
@@ -136,6 +142,7 @@ impl AsyncJobHandle {
#[cfg(test)]
mod tests {
use serde_json::json;
use time::{OffsetDateTime, format_description::well_known::Rfc3339};
use super::{AsyncJobHandle, JobStatus, StreamSession, StreamStatus};
use crate::{
@@ -144,6 +151,10 @@ mod tests {
streaming::ExecutionMode,
};
fn timestamp(value: &str) -> OffsetDateTime {
OffsetDateTime::parse(value, &Rfc3339).unwrap()
}
#[test]
fn stream_session_tracks_poll_and_close_transitions() {
let mut session = StreamSession {
@@ -156,20 +167,20 @@ mod tests {
status: StreamStatus::Running,
cursor: None,
state: json!({"cursor":"abc"}),
expires_at: "2026-04-06T12:05:00Z".to_owned(),
expires_at: timestamp("2026-04-06T12:05:00Z"),
last_poll_at: None,
created_at: "2026-04-06T12:00:00Z".to_owned(),
created_at: timestamp("2026-04-06T12:00:00Z"),
closed_at: None,
};
assert!(session.can_poll("2026-04-06T12:01:00Z"));
assert!(session.can_poll(timestamp("2026-04-06T12:01:00Z")));
session.mark_polled("2026-04-06T12:01:00Z");
session.mark_closed("2026-04-06T12:02:00Z");
session.mark_polled(timestamp("2026-04-06T12:01:00Z"));
session.mark_closed(timestamp("2026-04-06T12:02:00Z"));
assert_eq!(session.status, StreamStatus::Stopped);
assert_eq!(session.closed_at.as_deref(), Some("2026-04-06T12:02:00Z"));
assert!(!session.can_poll("2026-04-06T12:03:00Z"));
assert_eq!(session.closed_at, Some(timestamp("2026-04-06T12:02:00Z")));
assert!(!session.can_poll(timestamp("2026-04-06T12:03:00Z")));
}
#[test]
@@ -183,15 +194,15 @@ mod tests {
progress: json!({"percent": 60}),
result: None,
error: None,
expires_at: Some("2026-04-06T12:05:00Z".to_owned()),
created_at: "2026-04-06T12:00:00Z".to_owned(),
updated_at: "2026-04-06T12:00:00Z".to_owned(),
expires_at: Some(timestamp("2026-04-06T12:05:00Z")),
created_at: timestamp("2026-04-06T12:00:00Z"),
updated_at: timestamp("2026-04-06T12:00:00Z"),
finished_at: None,
};
assert!(job.can_cancel());
job.mark_finished("2026-04-06T12:01:00Z", json!({"ok": true}));
job.mark_finished(timestamp("2026-04-06T12:01:00Z"), json!({"ok": true}));
assert!(job.is_finished());
assert_eq!(job.status, JobStatus::Completed);
@@ -200,11 +211,14 @@ mod tests {
failed_job.result = None;
failed_job.finished_at = None;
failed_job.mark_failed("2026-04-06T12:02:00Z", json!({"message": "boom"}));
failed_job.mark_failed(
timestamp("2026-04-06T12:02:00Z"),
json!({"message": "boom"}),
);
assert_eq!(failed_job.status, JobStatus::Failed);
assert_eq!(
failed_job.finished_at.as_deref(),
Some("2026-04-06T12:02:00Z")
failed_job.finished_at,
Some(timestamp("2026-04-06T12:02:00Z"))
);
}
}
crates/crank-registry/src/model.rs
+7 -6
View File
@@ -10,6 +10,7 @@ use crank_mapping::MappingSet;
use crank_schema::Schema;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use time::OffsetDateTime;
macro_rules! define_registry_id {
($name:ident) => {
@@ -464,9 +465,9 @@ pub struct UpdateStreamSessionStateRequest<'a> {
pub next_status: StreamStatus,
pub cursor: Option<&'a Value>,
pub state: &'a Value,
pub expires_at: Option<&'a str>,
pub last_poll_at: Option<&'a str>,
pub closed_at: Option<&'a str>,
pub expires_at: Option<&'a OffsetDateTime>,
pub last_poll_at: Option<&'a OffsetDateTime>,
pub closed_at: Option<&'a OffsetDateTime>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -492,9 +493,9 @@ pub struct UpdateAsyncJobStatusRequest<'a> {
pub progress: &'a Value,
pub result: Option<&'a Value>,
pub error: Option<&'a Value>,
pub expires_at: Option<&'a str>,
pub updated_at: &'a str,
pub finished_at: Option<&'a str>,
pub expires_at: Option<&'a OffsetDateTime>,
pub updated_at: &'a OffsetDateTime,
pub finished_at: Option<&'a OffsetDateTime>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
crates/crank-registry/src/postgres/mod.rs
+19 -19
View File
@@ -2015,7 +2015,7 @@ mod tests {
.await
.unwrap();
registry
.close_stream_session(&closable_session.id, "2026-04-06T12:00:30Z")
.close_stream_session(&closable_session.id, &timestamp("2026-04-06T12:00:30Z"))
.await
.unwrap();
@@ -2033,14 +2033,14 @@ mod tests {
next_status: StreamStatus::Failed,
cursor: Some(&json!({"cursor":"next"})),
state: &json!({"phase":"failed"}),
expires_at: Some("2026-04-06T12:10:00Z"),
last_poll_at: Some("2026-04-06T12:01:00Z"),
closed_at: Some("2026-04-06T12:01:00Z"),
expires_at: Some(&timestamp("2026-04-06T12:10:00Z")),
last_poll_at: Some(&timestamp("2026-04-06T12:01:00Z")),
closed_at: Some(&timestamp("2026-04-06T12:01:00Z")),
})
.await
.unwrap();
assert_eq!(updated.status, StreamStatus::Failed);
assert_eq!(updated.closed_at.as_deref(), Some("2026-04-06T12:01:00Z"));
assert_eq!(updated.closed_at, Some(timestamp("2026-04-06T12:01:00Z")));
let page = registry
.list_stream_sessions(StreamSessionFilter {
@@ -2062,7 +2062,7 @@ mod tests {
.create_stream_session(CreateStreamSessionRequest {
session: &StreamSession {
id: expired_session.id.clone(),
expires_at: "2026-04-06T11:00:00Z".to_owned(),
expires_at: timestamp("2026-04-06T11:00:00Z"),
..expired_session
},
})
@@ -2070,7 +2070,7 @@ mod tests {
.unwrap();
let deleted = registry
.delete_expired_stream_sessions("2026-04-06T12:00:00Z")
.delete_expired_stream_sessions(&timestamp("2026-04-06T12:00:00Z"))
.await
.unwrap();
assert_eq!(deleted, 1);
@@ -2140,7 +2140,7 @@ mod tests {
.await
.unwrap();
registry
.cancel_async_job(&cancellable_job.id, "2026-04-06T12:00:30Z")
.cancel_async_job(&cancellable_job.id, &timestamp("2026-04-06T12:00:30Z"))
.await
.unwrap();
@@ -2159,9 +2159,9 @@ mod tests {
progress: &json!({"percent":100}),
result: Some(&json!({"ok":true})),
error: None,
expires_at: Some("2026-04-06T12:15:00Z"),
updated_at: "2026-04-06T12:01:00Z",
finished_at: Some("2026-04-06T12:01:00Z"),
expires_at: Some(&timestamp("2026-04-06T12:15:00Z")),
updated_at: &timestamp("2026-04-06T12:01:00Z"),
finished_at: Some(&timestamp("2026-04-06T12:01:00Z")),
})
.await
.unwrap();
@@ -2185,7 +2185,7 @@ mod tests {
let expired_job = AsyncJobHandle {
id: crank_core::AsyncJobId::new("job_expired"),
expires_at: Some("2026-04-06T11:00:00Z".to_owned()),
expires_at: Some(timestamp("2026-04-06T11:00:00Z")),
..test_async_job("job_expired", "op_job_01", JobStatus::Cancelled)
};
registry
@@ -2194,7 +2194,7 @@ mod tests {
.unwrap();
let deleted = registry
.delete_expired_async_jobs("2026-04-06T12:00:00Z")
.delete_expired_async_jobs(&timestamp("2026-04-06T12:00:00Z"))
.await
.unwrap();
assert_eq!(deleted, 1);
@@ -2227,7 +2227,7 @@ mod tests {
result: None,
error: None,
expires_at: None,
updated_at: "2026-04-06T12:01:00Z",
updated_at: &timestamp("2026-04-06T12:01:00Z"),
finished_at: None,
})
.await
@@ -2432,9 +2432,9 @@ mod tests {
status,
cursor: Some(json!({"cursor":"initial"})),
state: json!({"phase":"running"}),
expires_at: "2026-04-06T12:05:00Z".to_owned(),
expires_at: timestamp("2026-04-06T12:05:00Z"),
last_poll_at: None,
created_at: "2026-04-06T12:00:00Z".to_owned(),
created_at: timestamp("2026-04-06T12:00:00Z"),
closed_at: None,
}
}
@@ -2510,9 +2510,9 @@ mod tests {
progress: json!({"percent": 25}),
result: None,
error: None,
expires_at: Some("2026-04-06T12:05:00Z".to_owned()),
created_at: "2026-04-06T12:00:00Z".to_owned(),
updated_at: "2026-04-06T12:00:00Z".to_owned(),
expires_at: Some(timestamp("2026-04-06T12:05:00Z")),
created_at: timestamp("2026-04-06T12:00:00Z"),
updated_at: timestamp("2026-04-06T12:00:00Z"),
finished_at: None,
}
}
crates/crank-registry/src/postgres/stream.rs
+47 -36
View File
@@ -1,4 +1,5 @@
use super::*;
use time::OffsetDateTime;
impl PostgresRegistry {
pub async fn create_stream_session(
@@ -40,10 +41,10 @@ impl PostgresRegistry {
.bind(serialize_enum_text(&request.session.status, "status")?)
.bind(request.session.cursor.clone().map(Json))
.bind(Json(request.session.state.clone()))
.bind(&request.session.expires_at)
.bind(request.session.last_poll_at.as_deref())
.bind(&request.session.created_at)
.bind(request.session.closed_at.as_deref())
.bind(request.session.expires_at)
.bind(request.session.last_poll_at)
.bind(request.session.created_at)
.bind(request.session.closed_at)
.execute(&self.pool)
.await?;
@@ -65,10 +66,10 @@ impl PostgresRegistry {
status,
cursor_json,
state_json,
to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"expires_at!\",
to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at,
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"created_at!\",
to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at
expires_at as \"expires_at!: OffsetDateTime\",
last_poll_at as \"last_poll_at: OffsetDateTime\",
created_at as \"created_at!: OffsetDateTime\",
closed_at as \"closed_at: OffsetDateTime\"
from stream_sessions
where id = $1",
id.as_str(),
@@ -129,10 +130,10 @@ impl PostgresRegistry {
status,
cursor_json,
state_json,
to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at,
to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at,
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at",
expires_at,
last_poll_at,
created_at,
closed_at",
)
.bind(request.session_id.as_str())
.bind(serialize_enum_text(&request.current_status, "status")?)
@@ -166,7 +167,7 @@ impl PostgresRegistry {
pub async fn close_stream_session(
&self,
id: &StreamSessionId,
now: &str,
now: &OffsetDateTime,
) -> Result<(), RegistryError> {
let result = sqlx::query(
"update stream_sessions
@@ -224,10 +225,10 @@ impl PostgresRegistry {
status,
cursor_json,
state_json,
to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"expires_at!\",
to_char(last_poll_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as last_poll_at,
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"created_at!\",
to_char(closed_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as closed_at
expires_at as \"expires_at!: OffsetDateTime\",
last_poll_at as \"last_poll_at: OffsetDateTime\",
created_at as \"created_at!: OffsetDateTime\",
closed_at as \"closed_at: OffsetDateTime\"
from stream_sessions
where workspace_id = $1
and ($2::text is null or agent_id = $2)
@@ -271,7 +272,10 @@ impl PostgresRegistry {
Ok(Page { items, total })
}
pub async fn delete_expired_stream_sessions(&self, now: &str) -> Result<u64, RegistryError> {
pub async fn delete_expired_stream_sessions(
&self,
now: &OffsetDateTime,
) -> Result<u64, RegistryError> {
let result = sqlx::query(
"delete from stream_sessions
where expires_at <= $1::timestamptz",
@@ -314,10 +318,10 @@ impl PostgresRegistry {
.bind(Json(request.job.progress.clone()))
.bind(request.job.result.clone().map(Json))
.bind(request.job.error.clone().map(Json))
.bind(request.job.expires_at.as_deref())
.bind(&request.job.created_at)
.bind(&request.job.updated_at)
.bind(request.job.finished_at.as_deref())
.bind(request.job.expires_at)
.bind(request.job.created_at)
.bind(request.job.updated_at)
.bind(request.job.finished_at)
.execute(&self.pool)
.await?;
@@ -338,10 +342,10 @@ impl PostgresRegistry {
progress_json,
result_json,
error_json,
to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at,
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"created_at!\",
to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"updated_at!\",
to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at
expires_at as \"expires_at: OffsetDateTime\",
created_at as \"created_at!: OffsetDateTime\",
updated_at as \"updated_at!: OffsetDateTime\",
finished_at as \"finished_at: OffsetDateTime\"
from async_jobs
where id = $1",
id.as_str(),
@@ -397,10 +401,10 @@ impl PostgresRegistry {
progress_json,
result_json,
error_json,
to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at,
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as created_at,
to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as updated_at,
to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at",
expires_at,
created_at,
updated_at,
finished_at",
)
.bind(request.job_id.as_str())
.bind(serialize_enum_text(&request.current_status, "status")?)
@@ -432,7 +436,11 @@ impl PostgresRegistry {
}
}
pub async fn cancel_async_job(&self, id: &AsyncJobId, now: &str) -> Result<(), RegistryError> {
pub async fn cancel_async_job(
&self,
id: &AsyncJobId,
now: &OffsetDateTime,
) -> Result<(), RegistryError> {
let result = sqlx::query(
"update async_jobs
set status = 'cancelled',
@@ -483,10 +491,10 @@ impl PostgresRegistry {
progress_json,
result_json,
error_json,
to_char(expires_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as expires_at,
to_char(created_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"created_at!\",
to_char(updated_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as \"updated_at!\",
to_char(finished_at at time zone 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') as finished_at
expires_at as \"expires_at: OffsetDateTime\",
created_at as \"created_at!: OffsetDateTime\",
updated_at as \"updated_at!: OffsetDateTime\",
finished_at as \"finished_at: OffsetDateTime\"
from async_jobs
where workspace_id = $1
and ($2::text is null or agent_id = $2)
@@ -527,7 +535,10 @@ impl PostgresRegistry {
Ok(Page { items, total })
}
pub async fn delete_expired_async_jobs(&self, now: &str) -> Result<u64, RegistryError> {
pub async fn delete_expired_async_jobs(
&self,
now: &OffsetDateTime,
) -> Result<u64, RegistryError> {
let result = sqlx::query(
"delete from async_jobs
where expires_at is not null