obix 0.9.0

Implementation of outbox backed by PG / sqlx
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

//! The timer-driven partition maintainer job.
//!
//! Registered via
//! [`Outbox::register_partition_maintainer`](crate::out::Outbox::register_partition_maintainer),
//! it drives [`ensure_partitions`](super::ensure_partitions) on an internal
//! interval. The job stays resident and reschedules **only** when it falls out
//! of its loop — on shutdown (reschedule to resume after restart) or on an
//! `ensure_partitions` error (which propagates so the scheduler retries per the
//! configured [`RetrySettings`] — the alert). It carries no durable execution
//! state: each tick is idempotent off the current sequence head.

use async_trait::async_trait;
use serde::{Deserialize, Serialize};

use job::{
    CurrentJob, Job, JobType, ResidentJobCompletion, ResidentJobInitializer, ResidentJobRunner,
    RetrySettings,
};

use super::Partitions;
use crate::tables::MailboxTables;

/// Registration surface for the partition maintainer, mirroring
/// [`OutboxEventJobConfig`](crate::out::OutboxEventJobConfig): the job-scheduling
/// details (type + retry policy). The width / premake / poll-interval come from
/// the [`MailboxConfig`](crate::MailboxConfig) the outbox was initialised with.
#[derive(Clone)]
pub struct PartitionMaintainerConfig {
    pub job_type: JobType,
    pub retry_settings: RetrySettings,
}

impl PartitionMaintainerConfig {
    pub fn new(job_type: JobType) -> Self {
        Self {
            job_type,
            retry_settings: RetrySettings::repeat_indefinitely(),
        }
    }

    pub fn with_retry_settings(mut self, settings: RetrySettings) -> Self {
        self.retry_settings = settings;
        self
    }
}

/// The maintainer holds no durable execution state: each tick is idempotent
/// off the current sequence head, so nothing needs to survive a restart.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub(crate) struct PartitionMaintainerJobData {}

pub(crate) struct PartitionMaintainerJobInitializer<Tables>
where
    Tables: MailboxTables,
{
    partitions: Partitions<Tables>,
    job_type: JobType,
    retry_settings: RetrySettings,
    interval: std::time::Duration,
}

impl<Tables> PartitionMaintainerJobInitializer<Tables>
where
    Tables: MailboxTables,
{
    pub fn new(
        partitions: Partitions<Tables>,
        config: &PartitionMaintainerConfig,
        interval: std::time::Duration,
    ) -> Self {
        Self {
            partitions,
            job_type: config.job_type.clone(),
            retry_settings: config.retry_settings.clone(),
            interval,
        }
    }
}

impl<Tables> ResidentJobInitializer for PartitionMaintainerJobInitializer<Tables>
where
    Tables: MailboxTables,
{
    type Config = PartitionMaintainerJobData;

    fn job_type(&self) -> JobType {
        self.job_type.clone()
    }

    fn retry_on_error_settings(&self) -> RetrySettings {
        self.retry_settings.clone()
    }

    fn init(&self, _job: &Job) -> Result<Box<dyn ResidentJobRunner>, Box<dyn std::error::Error>> {
        Ok(Box::new(PartitionMaintainerJobRunner::<Tables> {
            partitions: self.partitions.clone(),
            interval: self.interval,
        }))
    }
}

struct PartitionMaintainerJobRunner<Tables>
where
    Tables: MailboxTables,
{
    partitions: Partitions<Tables>,
    interval: std::time::Duration,
}

#[async_trait]
impl<Tables> ResidentJobRunner for PartitionMaintainerJobRunner<Tables>
where
    Tables: MailboxTables,
{
    async fn run(
        &self,
        mut current_job: CurrentJob,
    ) -> Result<ResidentJobCompletion, Box<dyn std::error::Error>> {
        // Re-run on an internal timer rather than round-tripping through the job
        // scheduler each tick: the job stays resident and only falls out of the
        // loop on shutdown (reschedule to resume after restart) or on an
        // `ensure` error (propagated below — the scheduler then retries per
        // `retry_settings`, which is the alert).
        loop {
            self.partitions
                .ensure()
                .await
                .map_err(|e| Box::new(e) as Box<dyn std::error::Error>)?;

            tokio::select! {
                biased;
                _ = current_job.shutdown_requested() => {
                    return Ok(ResidentJobCompletion::RescheduleNow);
                }
                _ = tokio::time::sleep(self.interval) => {}
            }
        }
    }
}