1use std::collections::HashMap;
2use std::sync::{Arc, RwLock};
3use std::time::Instant;
4
5use chrono::{DateTime, Duration, Utc};
6use surrealdb::engine::any::Any;
7use surrealdb::Surreal;
8
9use crate::application::runtime::in_memory_runtime::{JobExecutionOutcome, JobHandler};
10use crate::application::runtime::replay_report::ReplayReport;
11use crate::application::runtime::retention::{RetentionPolicy, RetentionPruneReport};
12use crate::application::runtime::runtime_diagnostics_helpers;
13use crate::application::runtime::runtime_job_identity_context::RuntimeJobIdentityContext;
14use crate::application::use_cases::investigate_runtime_lineage::{
15 InvestigateRuntimeLineage, RuntimeLineageQuery, RuntimeLineageReport,
16};
17use crate::domain::errors::{Result, StasisError};
18use crate::domain::runtime::job::{JobState, NewJob};
19use crate::domain::runtime::job_attempt::{JobAttempt, JobAttemptOutcome};
20use crate::domain::runtime::outbox::{
21 OutboxEvent, OutboxPublishPolicy, OutboxStatus, RuntimeEvent, RuntimeEventType,
22};
23use crate::domain::runtime::recurring::RecurringDefinition;
24use crate::application::telemetry::keys as metric_keys;
25use crate::application::telemetry::operation::{runtime_event_type_name, OperationTelemetry};
26use crate::application::telemetry::propagation::{
27 job_execute_span_attributes, parent_trace_context,
28};
29use crate::application::telemetry::request_context::{
30 inbound_trace_context_for_propagation, trace_id_for_enqueue,
31};
32use crate::application::telemetry::spans as span_names;
33use crate::infrastructure::runtime::atomic_id_generator::AtomicIdGenerator;
34use crate::infrastructure::runtime::noop_runtime_metrics::NoopRuntimeMetrics;
35use crate::infrastructure::telemetry::NoopRuntimeTracing;
36use crate::infrastructure::runtime::surreal_job_attempt_store::SurrealJobAttemptStore;
37use crate::infrastructure::runtime::surreal_job_store::SurrealJobStore;
38use crate::infrastructure::runtime::surreal_outbox_store::SurrealOutboxStore;
39use crate::infrastructure::runtime::surreal_recurring_store::SurrealRecurringStore;
40use crate::infrastructure::runtime::system_clock::SystemClock;
41use crate::ports::outbound::runtime::clock::Clock;
42use crate::ports::outbound::runtime::event_publisher::EventPublisher;
43use crate::ports::outbound::runtime::id_generator::IdGenerator;
44use crate::ports::outbound::runtime::job_attempt_store::JobAttemptStore;
45use crate::ports::outbound::runtime::job_store::JobStore;
46use crate::ports::outbound::runtime::outbox_store::OutboxStore;
47use crate::ports::outbound::runtime::recurring_store::RecurringStore;
48use crate::ports::outbound::runtime::runtime_metrics::RuntimeMetrics;
49use crate::ports::outbound::runtime::runtime_tracing::{OtelAttribute, RuntimeTracing};
50
51#[derive(Clone)]
52pub struct SurrealRuntime {
53 pub job_store: SurrealJobStore,
54 pub recurring_store: SurrealRecurringStore,
55 pub outbox_store: SurrealOutboxStore,
56 pub job_attempt_store: SurrealJobAttemptStore,
57 handlers: Arc<RwLock<HashMap<String, Arc<dyn JobHandler>>>>,
58 publisher: Arc<RwLock<Option<Arc<dyn EventPublisher>>>>,
59 publish_policy: Arc<RwLock<OutboxPublishPolicy>>,
60 clock: Arc<dyn Clock>,
61 id_generator: Arc<dyn IdGenerator>,
62 metrics: Arc<dyn RuntimeMetrics>,
63 tracing: Arc<dyn RuntimeTracing>,
64 retention_policy: Arc<RwLock<RetentionPolicy>>,
65}
66
67impl SurrealRuntime {
68 pub fn new(db: Surreal<Any>) -> Self {
69 Self::with_dependencies_and_telemetry(
70 db,
71 Arc::new(SystemClock),
72 Arc::new(AtomicIdGenerator::new(1)),
73 Arc::new(NoopRuntimeMetrics),
74 Arc::new(NoopRuntimeTracing),
75 )
76 }
77
78 pub fn with_dependencies(
79 db: Surreal<Any>,
80 clock: Arc<dyn Clock>,
81 id_generator: Arc<dyn IdGenerator>,
82 ) -> Self {
83 Self::with_dependencies_and_telemetry(
84 db,
85 clock,
86 id_generator,
87 Arc::new(NoopRuntimeMetrics),
88 Arc::new(NoopRuntimeTracing),
89 )
90 }
91
92 pub fn with_dependencies_and_metrics(
93 db: Surreal<Any>,
94 clock: Arc<dyn Clock>,
95 id_generator: Arc<dyn IdGenerator>,
96 metrics: Arc<dyn RuntimeMetrics>,
97 ) -> Self {
98 Self::with_dependencies_and_telemetry(db, clock, id_generator, metrics, Arc::new(NoopRuntimeTracing))
99 }
100
101 pub fn with_dependencies_and_telemetry(
102 db: Surreal<Any>,
103 clock: Arc<dyn Clock>,
104 id_generator: Arc<dyn IdGenerator>,
105 metrics: Arc<dyn RuntimeMetrics>,
106 tracing: Arc<dyn RuntimeTracing>,
107 ) -> Self {
108 Self {
109 job_store: SurrealJobStore::new(db.clone()),
110 recurring_store: SurrealRecurringStore::new(db.clone()),
111 outbox_store: SurrealOutboxStore::new(db.clone()),
112 job_attempt_store: SurrealJobAttemptStore::new(db),
113 handlers: Arc::new(RwLock::new(HashMap::new())),
114 publisher: Arc::new(RwLock::new(None)),
115 publish_policy: Arc::new(RwLock::new(OutboxPublishPolicy::default())),
116 clock,
117 id_generator,
118 metrics,
119 tracing,
120 retention_policy: Arc::new(RwLock::new(RetentionPolicy::default())),
121 }
122 }
123
124 pub fn replace_telemetry(
125 &mut self,
126 metrics: Arc<dyn RuntimeMetrics>,
127 tracing: Arc<dyn RuntimeTracing>,
128 ) {
129 self.metrics = metrics;
130 self.tracing = tracing;
131 }
132
133 pub fn tracing(&self) -> Arc<dyn RuntimeTracing> {
134 self.tracing.clone()
135 }
136
137 pub fn configure_retention_policy(&self, policy: RetentionPolicy) -> Result<()> {
138 let mut state = self
139 .retention_policy
140 .write()
141 .map_err(|_| StasisError::PortFailure("retention policy lock poisoned".to_string()))?;
142 *state = policy;
143 Ok(())
144 }
145
146 pub fn register_handler<H: JobHandler + 'static>(&self, handler: H) -> Result<()> {
147 let mut handlers = self
148 .handlers
149 .write()
150 .map_err(|_| StasisError::PortFailure("handlers lock poisoned".to_string()))?;
151
152 handlers.insert(handler.job_type().to_string(), Arc::new(handler));
153 Ok(())
154 }
155
156 pub fn register_event_publisher<P: EventPublisher + 'static>(
157 &self,
158 publisher: P,
159 ) -> Result<()> {
160 let mut state = self
161 .publisher
162 .write()
163 .map_err(|_| StasisError::PortFailure("publisher lock poisoned".to_string()))?;
164
165 *state = Some(Arc::new(publisher));
166 Ok(())
167 }
168
169 pub fn configure_outbox_publish_policy(&self, policy: OutboxPublishPolicy) -> Result<()> {
170 let mut state = self
171 .publish_policy
172 .write()
173 .map_err(|_| StasisError::PortFailure("publish policy lock poisoned".to_string()))?;
174
175 *state = policy;
176 Ok(())
177 }
178
179 pub async fn enqueue(&self, job: NewJob) -> Result<()> {
180 self.job_store.insert(job.into_job()).await
181 }
182
183 pub async fn register_recurring(&self, definition: RecurringDefinition) -> Result<()> {
184 self.recurring_store.insert(definition).await
185 }
186
187 pub async fn list_job_attempts(&self, job_id: &str) -> Result<Vec<JobAttempt>> {
188 self.job_attempt_store.list_by_job_id(job_id).await
189 }
190
191 pub async fn list_attempts_by_guardrail_code(
192 &self,
193 guardrail_code: &str,
194 ) -> Result<Vec<JobAttempt>> {
195 self.job_attempt_store
196 .list_by_guardrail_code(guardrail_code)
197 .await
198 }
199
200 pub async fn list_attempts_by_execution_id(
201 &self,
202 execution_id: &str,
203 ) -> Result<Vec<JobAttempt>> {
204 self.job_attempt_store
205 .list_by_execution_id(execution_id)
206 .await
207 }
208
209 pub async fn list_lineage_events(&self, job_id: &str) -> Result<Vec<OutboxEvent>> {
210 self.outbox_store.list_by_job_id(job_id).await
211 }
212
213 pub async fn list_lineage_events_by_execution_id(
214 &self,
215 execution_id: &str,
216 ) -> Result<Vec<OutboxEvent>> {
217 self.outbox_store.list_by_execution_id(execution_id).await
218 }
219
220 pub async fn list_lineage_events_by_thread_id(
221 &self,
222 thread_id: &str,
223 ) -> Result<Vec<OutboxEvent>> {
224 self.outbox_store.list_by_thread_id(thread_id).await
225 }
226
227 pub async fn investigate_lineage(
228 &self,
229 query: RuntimeLineageQuery,
230 ) -> Result<RuntimeLineageReport> {
231 InvestigateRuntimeLineage::new(self.job_attempt_store.clone(), self.outbox_store.clone())
232 .execute(query)
233 .await
234 }
235
236 pub async fn get_replay_report(&self, job_id: &str) -> Result<ReplayReport> {
237 Ok(ReplayReport {
238 job_id: job_id.to_string(),
239 attempts: self.list_job_attempts(job_id).await?,
240 lineage_events: self.list_lineage_events(job_id).await?,
241 })
242 }
243
244 pub async fn process_once_now(&self, queue: &str, worker_id: &str) -> Result<Option<String>> {
245 self.process_once(queue, worker_id, self.clock.now()).await
246 }
247
248 pub async fn replay_dead_letter_now(&self, job_id: &str) -> Result<bool> {
249 self.replay_dead_letter(job_id, self.clock.now()).await
250 }
251
252 pub async fn publish_pending_events_now(&self, limit: usize) -> Result<usize> {
253 self.publish_pending_events(limit, self.clock.now()).await
254 }
255
256 pub async fn materialize_recurring_now(&self, scheduler_id: &str) -> Result<usize> {
257 self.materialize_recurring(self.clock.now(), scheduler_id)
258 .await
259 }
260
261 pub async fn prune_terminal_records(
262 &self,
263 cutoff: DateTime<Utc>,
264 ) -> Result<RetentionPruneReport> {
265 Ok(RetentionPruneReport {
266 jobs_pruned: self.job_store.prune_terminal_before(cutoff).await?,
267 attempts_pruned: self.job_attempt_store.prune_finished_before(cutoff).await?,
268 outbox_events_pruned: self.outbox_store.prune_non_pending_before(cutoff).await?,
269 })
270 }
271
272 pub async fn enforce_retention(&self, now: DateTime<Utc>) -> Result<RetentionPruneReport> {
273 let policy = self
274 .retention_policy
275 .read()
276 .map_err(|_| StasisError::PortFailure("retention policy lock poisoned".to_string()))?
277 .clone();
278 let cutoff = now - Duration::days(policy.terminal_ttl_days.max(0));
279 self.prune_terminal_records(cutoff).await
280 }
281
282 pub async fn enforce_retention_now(&self) -> Result<RetentionPruneReport> {
283 self.enforce_retention(self.clock.now()).await
284 }
285
286 pub async fn materialize_recurring(
287 &self,
288 now: DateTime<Utc>,
289 scheduler_id: &str,
290 ) -> Result<usize> {
291 let due = self
292 .recurring_store
293 .lease_due(now, scheduler_id, 30)
294 .await?;
295
296 let mut produced = 0usize;
297
298 for mut definition in due {
299 if !definition.enabled {
300 continue;
301 }
302
303 let id = self.id_generator.next_id(&definition.id).to_string();
304
305 let scheduled_at = now + Duration::seconds(definition.jitter_seconds.max(0));
306
307 let job = NewJob {
308 id,
309 queue: definition.queue.clone(),
310 job_type: definition.job_type.clone(),
311 payload_ref: definition.payload_template_ref.clone(),
312 priority: 100,
313 max_attempts: definition.max_attempts,
314 idempotency_key: format!("recurring:{}:{}", definition.id, now.timestamp()),
315 correlation_id: definition.id.clone(),
316 causation_id: definition.id.clone(),
317 trace_id: trace_id_for_enqueue(|| definition.id.clone()),
318 sttp_input_node_id: definition.payload_template_ref.clone(),
319 scheduled_at,
320 backoff_policy: Default::default(),
321 };
322
323 self.enqueue(job).await?;
324
325 definition.last_run_at = Some(now);
326 definition.next_run_at = definition.compute_next_run_at(now)?;
327 definition.lease_owner = None;
328 definition.lease_expires_at = None;
329 self.recurring_store.save(definition).await?;
330 produced += 1;
331 }
332
333 Ok(produced)
334 }
335
336 pub async fn process_once(
337 &self,
338 queue: &str,
339 worker_id: &str,
340 now: DateTime<Utc>,
341 ) -> Result<Option<String>> {
342 let worker_started = Instant::now();
343 let worker_parent = inbound_trace_context_for_propagation();
344 let _worker_span = self.tracing.start_span_with_trace_context(
345 span_names::WORKER_PROCESS_ONCE,
346 &[
347 OtelAttribute::string("stasis.queue", queue.to_string()),
348 OtelAttribute::string("stasis.worker_id", worker_id.to_string()),
349 ],
350 worker_parent.as_ref(),
351 );
352 self.metrics
353 .incr_counter(metric_keys::WORKER_PROCESS_ONCE_TOTAL, 1);
354
355 let Some(mut job) = self.job_store.lease_due(queue, worker_id, now, 30).await? else {
356 self.metrics.observe_duration_ms(
357 metric_keys::WORKER_PROCESS_ONCE_DURATION_MS,
358 worker_started.elapsed().as_millis() as u64,
359 );
360 return Ok(None);
361 };
362
363 job.state = JobState::Running;
364 job.started_at = job.started_at.or(Some(now));
365 job.heartbeat_at = Some(now);
366 let job_identity = RuntimeJobIdentityContext::from(&job);
367 self.job_store.save(job.clone()).await?;
368 let processing_started = Instant::now();
369
370 let handler = {
371 let handlers = self
372 .handlers
373 .read()
374 .map_err(|_| StasisError::PortFailure("handlers lock poisoned".to_string()))?;
375 handlers.get(&job.job_type).cloned()
376 };
377
378 let job_parent = parent_trace_context(&job.trace_id)
379 .or_else(inbound_trace_context_for_propagation);
380 let _job_span = self.tracing.start_span_with_trace_context(
381 span_names::JOB_EXECUTE,
382 &job_execute_span_attributes(&job),
383 job_parent.as_ref(),
384 );
385
386 let outcome = if let Some(handler) = handler {
387 handler.execute(&job).await?
388 } else {
389 JobExecutionOutcome::FatalFailure {
390 message: format!("no handler registered for job_type={}", job.job_type),
391 execution_id: None,
392 diagnostics: None,
393 }
394 };
395
396 let attempt_number = job.attempts + 1;
397 let attempt_started_at = now;
398
399 match outcome {
400 JobExecutionOutcome::Success {
401 sttp_output_node_id,
402 execution_id,
403 diagnostics,
404 } => {
405 let diagnostics_envelope =
406 Self::extract_diagnostics_envelope(diagnostics.as_deref());
407 job.state = JobState::Succeeded;
408 job.sttp_output_node_id = Some(sttp_output_node_id.clone());
409 job.finished_at = Some(now);
410 job.lease_owner = None;
411 job.lease_expires_at = None;
412 job.heartbeat_at = None;
413 self.job_store.save(job).await?;
414
415 self.append_outbox(
416 RuntimeEventType::JobSucceeded,
417 &job_identity,
418 Some(sttp_output_node_id.clone()),
419 None,
420 now,
421 execution_id.clone(),
422 &diagnostics_envelope,
423 )
424 .await?;
425
426 self.append_job_attempt(
427 &job_identity.job_id,
428 worker_id,
429 attempt_number,
430 attempt_started_at,
431 now,
432 JobAttemptOutcome::Succeeded,
433 None,
434 Some(sttp_output_node_id),
435 execution_id,
436 &diagnostics_envelope,
437 diagnostics,
438 )
439 .await?;
440
441 self.metrics.incr_counter(metric_keys::JOB_SUCCEEDED_TOTAL, 1);
442 self.metrics.observe_duration_ms(
443 metric_keys::JOB_PROCESS_DURATION_MS,
444 processing_started.elapsed().as_millis() as u64,
445 );
446 }
447 JobExecutionOutcome::RetryableFailure {
448 message,
449 execution_id,
450 diagnostics,
451 } => {
452 let diagnostics_envelope =
453 Self::extract_diagnostics_envelope(diagnostics.as_deref());
454 let guardrail_failure = diagnostics
455 .as_deref()
456 .map(|v| v.contains("\"guardrail_code\""))
457 .unwrap_or(false);
458 job.attempts += 1;
459 job.last_error = Some(message.clone());
460 job.lease_owner = None;
461 job.lease_expires_at = None;
462 job.heartbeat_at = None;
463
464 if job.attempts >= job.max_attempts {
465 job.state = JobState::DeadLetter;
466 job.finished_at = Some(now);
467 self.append_outbox(
468 RuntimeEventType::JobDeadLettered,
469 &job_identity,
470 None,
471 Some(message.clone()),
472 now,
473 execution_id.clone(),
474 &diagnostics_envelope,
475 )
476 .await?;
477
478 self.metrics.incr_counter(metric_keys::JOB_DEAD_LETTER_TOTAL, 1);
479 } else {
480 job.state = JobState::Enqueued;
481 let exponent = job.attempts - 1;
482 let mut delay = job
483 .backoff_policy
484 .base_delay_seconds
485 .saturating_mul(2_i64.saturating_pow(exponent));
486 delay = delay.min(job.backoff_policy.max_delay_seconds);
487 job.scheduled_at = now + Duration::seconds(delay.max(0));
488
489 self.append_outbox(
490 RuntimeEventType::JobRetryScheduled,
491 &job_identity,
492 None,
493 Some(message.clone()),
494 now,
495 execution_id.clone(),
496 &diagnostics_envelope,
497 )
498 .await?;
499
500 self.metrics
501 .incr_counter(metric_keys::JOB_RETRY_SCHEDULED_TOTAL, 1);
502 }
503
504 self.job_store.save(job).await?;
505
506 self.append_job_attempt(
507 &job_identity.job_id,
508 worker_id,
509 attempt_number,
510 attempt_started_at,
511 now,
512 JobAttemptOutcome::RetryableFailure,
513 Some(message),
514 None,
515 execution_id,
516 &diagnostics_envelope,
517 diagnostics,
518 )
519 .await?;
520
521 self.metrics
522 .incr_counter(metric_keys::JOB_RETRYABLE_FAILURE_TOTAL, 1);
523 self.metrics.observe_duration_ms(
524 metric_keys::JOB_PROCESS_DURATION_MS,
525 processing_started.elapsed().as_millis() as u64,
526 );
527 if guardrail_failure {
528 self.metrics
529 .incr_counter(metric_keys::GRAPHEME_GUARDRAIL_FAILURE_TOTAL, 1);
530 }
531 }
532 JobExecutionOutcome::FatalFailure {
533 message,
534 execution_id,
535 diagnostics,
536 } => {
537 let diagnostics_envelope =
538 Self::extract_diagnostics_envelope(diagnostics.as_deref());
539 let guardrail_failure = diagnostics
540 .as_deref()
541 .map(|v| v.contains("\"guardrail_code\""))
542 .unwrap_or(false);
543 job.attempts += 1;
544 job.state = JobState::DeadLetter;
545 job.last_error = Some(message.clone());
546 job.finished_at = Some(now);
547 job.lease_owner = None;
548 job.lease_expires_at = None;
549 job.heartbeat_at = None;
550 self.job_store.save(job).await?;
551
552 self.append_outbox(
553 RuntimeEventType::JobDeadLettered,
554 &job_identity,
555 None,
556 Some(message.clone()),
557 now,
558 execution_id.clone(),
559 &diagnostics_envelope,
560 )
561 .await?;
562
563 self.append_job_attempt(
564 &job_identity.job_id,
565 worker_id,
566 attempt_number,
567 attempt_started_at,
568 now,
569 JobAttemptOutcome::FatalFailure,
570 Some(message),
571 None,
572 execution_id,
573 &diagnostics_envelope,
574 diagnostics,
575 )
576 .await?;
577
578 self.metrics.incr_counter(metric_keys::JOB_FATAL_FAILURE_TOTAL, 1);
579 self.metrics.incr_counter(metric_keys::JOB_DEAD_LETTER_TOTAL, 1);
580 self.metrics.observe_duration_ms(
581 metric_keys::JOB_PROCESS_DURATION_MS,
582 processing_started.elapsed().as_millis() as u64,
583 );
584 if guardrail_failure {
585 self.metrics
586 .incr_counter(metric_keys::GRAPHEME_GUARDRAIL_FAILURE_TOTAL, 1);
587 }
588 }
589 }
590
591 self.metrics.observe_duration_ms(
592 metric_keys::WORKER_PROCESS_ONCE_DURATION_MS,
593 worker_started.elapsed().as_millis() as u64,
594 );
595 Ok(Some(job_identity.job_id))
596 }
597
598 pub async fn replay_dead_letter(&self, job_id: &str, now: DateTime<Utc>) -> Result<bool> {
599 let Some(mut job) = self.job_store.get(job_id).await? else {
600 return Ok(false);
601 };
602
603 if job.state != JobState::DeadLetter {
604 return Ok(false);
605 }
606
607 job.state = JobState::Enqueued;
608 job.attempts = 0;
609 job.last_error = None;
610 job.scheduled_at = now;
611 job.lease_owner = None;
612 job.lease_expires_at = None;
613 job.heartbeat_at = None;
614 job.finished_at = None;
615
616 self.job_store.save(job).await?;
617 Ok(true)
618 }
619
620 pub async fn publish_pending_events(&self, limit: usize, now: DateTime<Utc>) -> Result<usize> {
621 let publisher = {
622 let state = self
623 .publisher
624 .read()
625 .map_err(|_| StasisError::PortFailure("publisher lock poisoned".to_string()))?;
626 state.clone()
627 };
628
629 let Some(publisher) = publisher else {
630 return Ok(0);
631 };
632
633 let policy = self
634 .publish_policy
635 .read()
636 .map_err(|_| StasisError::PortFailure("publish policy lock poisoned".to_string()))?
637 .clone();
638
639 let pending = self.outbox_store.list_pending(limit).await?;
640 let mut published = 0usize;
641 let operation_telemetry = OperationTelemetry::new(self.metrics.clone(), self.tracing.clone());
642
643 for mut event in pending {
644 if event
645 .next_attempt_at
646 .map(|next| next > now)
647 .unwrap_or(false)
648 {
649 continue;
650 }
651
652 let event_type = runtime_event_type_name(&event.event.event_type);
653 let _publish_span = operation_telemetry.outbox_publish_span(event_type, &event.event.job_id);
654
655 match publisher.publish(&event).await {
656 Ok(()) => {
657 event.status = OutboxStatus::Published;
658 event.publish_attempts = event.publish_attempts.saturating_add(1);
659 event.published_at = Some(now);
660 event.next_attempt_at = None;
661 event.last_publish_error = None;
662 self.outbox_store.save(event).await?;
663 published += 1;
664 self.metrics
665 .incr_counter(metric_keys::OUTBOX_PUBLISH_SUCCESS_TOTAL, 1);
666 }
667 Err(err) => {
668 event.publish_attempts = event.publish_attempts.saturating_add(1);
669 event.published_at = None;
670 event.last_publish_error = Some(err.to_string());
671
672 if event.publish_attempts >= policy.max_attempts {
673 event.status = OutboxStatus::Failed;
674 event.next_attempt_at = None;
675 } else {
676 let exponent = event.publish_attempts - 1;
677 let mut delay = policy
678 .base_delay_seconds
679 .saturating_mul(2_i64.saturating_pow(exponent));
680 delay = delay.min(policy.max_delay_seconds);
681 event.status = OutboxStatus::Pending;
682 event.next_attempt_at = Some(now + Duration::seconds(delay.max(0)));
683 }
684
685 self.outbox_store.save(event).await?;
686 self.metrics
687 .incr_counter(metric_keys::OUTBOX_PUBLISH_FAILURE_TOTAL, 1);
688 }
689 }
690 }
691
692 Ok(published)
693 }
694
695 #[allow(clippy::too_many_arguments)]
696 async fn append_outbox(
697 &self,
698 event_type: RuntimeEventType,
699 job_identity: &RuntimeJobIdentityContext,
700 sttp_output_node_id: Option<String>,
701 message: Option<String>,
702 now: DateTime<Utc>,
703 execution_id: Option<String>,
704 diagnostics: &runtime_diagnostics_helpers::RuntimeDiagnosticsEnvelope,
705 ) -> Result<()> {
706 let event = OutboxEvent {
707 event_id: self
708 .id_generator
709 .next_id(&format!("evt-{}", job_identity.job_id)),
710 status: OutboxStatus::Pending,
711 publish_attempts: 0,
712 published_at: None,
713 next_attempt_at: None,
714 last_publish_error: None,
715 event: RuntimeEvent {
716 event_type,
717 job_id: job_identity.job_id.clone(),
718 thread_id: diagnostics.thread_id.clone(),
719 correlation_id: job_identity.correlation_id.clone(),
720 causation_id: job_identity.causation_id.clone(),
721 trace_id: job_identity.trace_id.clone(),
722 sttp_input_node_id: job_identity.sttp_input_node_id.clone(),
723 sttp_output_node_id,
724 execution_id,
725 input_memory_query_id: diagnostics.input_memory_query_id.clone(),
726 input_memory_query_fingerprint: diagnostics
727 .input_memory_query_fingerprint
728 .clone(),
729 output_memory_node_id: diagnostics.output_memory_node_id.clone(),
730 retrieval_path: diagnostics.retrieval_path.clone(),
731 occurred_at: now,
732 message,
733 },
734 };
735
736 self.outbox_store.insert(event).await
737 }
738
739 #[allow(clippy::too_many_arguments)]
740 async fn append_job_attempt(
741 &self,
742 job_id: &str,
743 worker_id: &str,
744 attempt_number: u32,
745 started_at: DateTime<Utc>,
746 finished_at: DateTime<Utc>,
747 outcome: JobAttemptOutcome,
748 error_message: Option<String>,
749 sttp_output_node_id: Option<String>,
750 execution_id: Option<String>,
751 diagnostics_envelope: &runtime_diagnostics_helpers::RuntimeDiagnosticsEnvelope,
752 diagnostics: Option<String>,
753 ) -> Result<()> {
754 let attempt = JobAttempt {
755 attempt_id: self.id_generator.next_id(&format!("attempt-{job_id}")),
756 job_id: job_id.to_string(),
757 attempt_number,
758 worker_id: worker_id.to_string(),
759 started_at,
760 finished_at,
761 outcome,
762 error_message,
763 sttp_output_node_id,
764 execution_id,
765 guardrail_code: diagnostics_envelope.guardrail_code.clone(),
766 policy_reason: diagnostics_envelope.policy_reason.clone(),
767 duration_ms: diagnostics_envelope.duration_ms,
768 diagnostics,
769 };
770
771 self.job_attempt_store.insert(attempt).await
772 }
773
774 fn extract_diagnostics_envelope(
775 diagnostics: Option<&str>,
776 ) -> runtime_diagnostics_helpers::RuntimeDiagnosticsEnvelope {
777 runtime_diagnostics_helpers::extract_runtime_diagnostics_envelope(diagnostics)
778 }
779
780}