24391673810

Committed 14 Apr 2026 09:31AM UTC coverage: 78.406% (+0.03%) from 78.38%

Build # 24391673810

Build Type

Pull #668

github

Committed by

web-flow

Commit Message

Merge 1aa184387 into 3065d5779

Pull Request Pull Request #668: refactor(etl/store): copy-on-write instead of clonning stored state

Coverage Stats

47 of 63 new or added lines in 6 files covered. (74.6%)

5 existing lines in 3 files now uncovered.

24312 of 31008 relevant lines covered (78.41%)

1083.13 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/etl-replicator/src/error_reporting.rs

use etl::error::{EtlError, EtlResult};
use etl::state::table::TableReplicationPhase;
use etl::store::cleanup::CleanupStore;
use etl::store::schema::SchemaStore;
use etl::store::state::{StateStore, TableMappings, TableReplicationStates};
use etl::types::{TableId, TableSchema};
use std::sync::Arc;
use tracing::info;

use crate::error_notification::ErrorNotificationClient;
use crate::sentry;

/// State store decorator that reports persisted table replication errors.
///
/// After [`StateStore::update_table_replication_states`] succeeds, this wrapper
/// reports each [`TableReplicationPhase::Errored`] update to Sentry and, when
/// configured, to the Supabase error-notification endpoint.
#[derive(Debug, Clone)]
pub struct ErrorReportingStateStore<S> {
    inner: S,
    notification_client: Option<Arc<ErrorNotificationClient>>,
}

/// Persisted table error waiting to be reported.
#[derive(Debug)]
struct ReportableTableError {
    /// Table whose replication state was persisted as errored.
    table_id: TableId,
    /// Source ETL error captured in the persisted table phase.
    source_err: EtlError,
}

impl<S> ErrorReportingStateStore<S> {
    /// Creates a reporting wrapper around `inner`.
    pub fn new(inner: S, notification_client: Option<ErrorNotificationClient>) -> Self {
        Self {
            inner,
            notification_client: notification_client.map(Arc::new),
        }
    }

    /// Reports persisted errored table state updates.
    async fn report_errored_updates(&self, updates: Vec<ReportableTableError>) {
        let notification_client = self.notification_client.as_ref();

        for update in updates {
            info!(
                table_id = update.table_id.0,
                "reporting table replication error"
            );

            sentry::capture_table_error(update.table_id, &update.source_err);
            if let Some(notification_client) = notification_client {
                notification_client
                    .notify_error(update.source_err.to_string(), &update.source_err)
                    .await;
            }
        }
    }

    /// Extracts only the errored state updates that need post-persistence reporting.
    fn collect_reportable_errors(
        updates: &[(TableId, TableReplicationPhase)],
    ) -> Vec<ReportableTableError> {
        updates
            .iter()
            .filter_map(|(table_id, phase)| match phase {
                TableReplicationPhase::Errored { source_err, .. } => Some(ReportableTableError {
                    table_id: *table_id,
                    source_err: source_err.clone(),
                }),
                _ => None,
            })
            .collect()
    }
}

impl<S> StateStore for ErrorReportingStateStore<S>
where
    S: StateStore + Send + Sync,
{
    async fn get_table_replication_state(
        &self,
        table_id: TableId,
    ) -> EtlResult<Option<TableReplicationPhase>> {
        self.inner.get_table_replication_state(table_id).await
    }

    async fn get_table_replication_states(&self) -> EtlResult<TableReplicationStates> {
        self.inner.get_table_replication_states().await
    }

    async fn load_table_replication_states(&self) -> EtlResult<usize> {
        self.inner.load_table_replication_states().await
    }

    async fn update_table_replication_states(
        &self,
        updates: Vec<(TableId, TableReplicationPhase)>,
    ) -> EtlResult<()> {
        // We collect all errors in advance, to avoid cloning the whole set of updates.
        let reportable_errors = Self::collect_reportable_errors(&updates);

        self.inner.update_table_replication_states(updates).await?;

        // This operation must be infallible or at least not propagate failures, otherwise the
        // error thrown here, will be caught and handled by the core of etl itself. There is no
        // infinite recursion problem, but it might make the system harder to understand.
        self.report_errored_updates(reportable_errors).await;

        Ok(())
    }

    async fn rollback_table_replication_state(
        &self,
        table_id: TableId,
    ) -> EtlResult<TableReplicationPhase> {
        self.inner.rollback_table_replication_state(table_id).await
    }

    async fn get_table_mapping(&self, source_table_id: &TableId) -> EtlResult<Option<String>> {
        self.inner.get_table_mapping(source_table_id).await
    }

    async fn get_table_mappings(&self) -> EtlResult<TableMappings> {
        self.inner.get_table_mappings().await
    }

    async fn load_table_mappings(&self) -> EtlResult<usize> {
        self.inner.load_table_mappings().await
    }

    async fn store_table_mapping(
        &self,
        source_table_id: TableId,
        destination_table_id: String,
    ) -> EtlResult<()> {
        self.inner
            .store_table_mapping(source_table_id, destination_table_id)
            .await
    }
}

impl<S> SchemaStore for ErrorReportingStateStore<S>
where
    S: SchemaStore + Send + Sync,
{
    async fn get_table_schema(&self, table_id: &TableId) -> EtlResult<Option<Arc<TableSchema>>> {
        self.inner.get_table_schema(table_id).await
    }

    async fn get_table_schemas(&self) -> EtlResult<Vec<Arc<TableSchema>>> {
        self.inner.get_table_schemas().await
    }

    async fn load_table_schemas(&self) -> EtlResult<usize> {
        self.inner.load_table_schemas().await
    }

    async fn store_table_schema(&self, table_schema: TableSchema) -> EtlResult<Arc<TableSchema>> {
        self.inner.store_table_schema(table_schema).await
    }
}

impl<S> CleanupStore for ErrorReportingStateStore<S>
where
    S: CleanupStore + Send + Sync,
{
    async fn cleanup_table_state(&self, table_id: TableId) -> EtlResult<()> {
        self.inner.cleanup_table_state(table_id).await
    }
}

1	use etl::error::{EtlError, EtlResult};
2	use etl::state::table::TableReplicationPhase;
3	use etl::store::cleanup::CleanupStore;
4	use etl::store::schema::SchemaStore;
5	use etl::store::state::{StateStore, TableMappings, TableReplicationStates};
6	use etl::types::{TableId, TableSchema};
7	use std::sync::Arc;
8	use tracing::info;
9
10	use crate::error_notification::ErrorNotificationClient;
11	use crate::sentry;
12
13	/// State store decorator that reports persisted table replication errors.
14	///
15	/// After [`StateStore::update_table_replication_states`] succeeds, this wrapper
16	/// reports each [`TableReplicationPhase::Errored`] update to Sentry and, when
17	/// configured, to the Supabase error-notification endpoint.
18	#[derive(Debug, Clone)]
19	pub struct ErrorReportingStateStore<S> {
20	inner: S,
21	notification_client: Option<Arc<ErrorNotificationClient>>,
22	}
23
24	/// Persisted table error waiting to be reported.
25	#[derive(Debug)]
26	struct ReportableTableError {
27	/// Table whose replication state was persisted as errored.
28	table_id: TableId,
29	/// Source ETL error captured in the persisted table phase.
30	source_err: EtlError,
31	}
32
33	impl<S> ErrorReportingStateStore<S> {
34	/// Creates a reporting wrapper around `inner`.
35	pub fn new(inner: S, notification_client: Option<ErrorNotificationClient>) -> Self {	×
36	Self {	×
37	inner,	×
38	notification_client: notification_client.map(Arc::new),	×
39	}	×
40	}	×
41
42	/// Reports persisted errored table state updates.
43	async fn report_errored_updates(&self, updates: Vec<ReportableTableError>) {	×
44	let notification_client = self.notification_client.as_ref();	×
45
46	for update in updates {	×
47	info!(	×
48	table_id = update.table_id.0,
49	"reporting table replication error"
50	);
51
52	sentry::capture_table_error(update.table_id, &update.source_err);	×
53	if let Some(notification_client) = notification_client {	×
54	notification_client	×
55	.notify_error(update.source_err.to_string(), &update.source_err)	×
56	.await;	×
57	}	×
58	}
59	}	×
60
61	/// Extracts only the errored state updates that need post-persistence reporting.
62	fn collect_reportable_errors(	×
63	updates: &[(TableId, TableReplicationPhase)],	×
64	) -> Vec<ReportableTableError> {	×
65	updates	×
66	.iter()	×
67	.filter_map(\|(table_id, phase)\| match phase {	×
68	TableReplicationPhase::Errored { source_err, .. } => Some(ReportableTableError {	×
69	table_id: *table_id,	×
70	source_err: source_err.clone(),	×
71	}),	×
72	_ => None,	×
73	})	×
74	.collect()	×
75	}	×
76	}
77
78	impl<S> StateStore for ErrorReportingStateStore<S>
79	where
80	S: StateStore + Send + Sync,
81	{
82	async fn get_table_replication_state(	×
83	&self,	×
84	table_id: TableId,	×
85	) -> EtlResult<Option<TableReplicationPhase>> {	×
86	self.inner.get_table_replication_state(table_id).await	×
87	}	×
88
NEW 89	async fn get_table_replication_states(&self) -> EtlResult<TableReplicationStates> {	×
90	self.inner.get_table_replication_states().await	×
91	}	×
92
93	async fn load_table_replication_states(&self) -> EtlResult<usize> {	×
94	self.inner.load_table_replication_states().await	×
95	}	×
96
97	async fn update_table_replication_states(	×
98	&self,	×
99	updates: Vec<(TableId, TableReplicationPhase)>,	×
100	) -> EtlResult<()> {	×
101	// We collect all errors in advance, to avoid cloning the whole set of updates.
102	let reportable_errors = Self::collect_reportable_errors(&updates);	×
103
104	self.inner.update_table_replication_states(updates).await?;	×
105
106	// This operation must be infallible or at least not propagate failures, otherwise the
107	// error thrown here, will be caught and handled by the core of etl itself. There is no
108	// infinite recursion problem, but it might make the system harder to understand.
109	self.report_errored_updates(reportable_errors).await;	×
110
111	Ok(())	×
112	}	×
113
114	async fn rollback_table_replication_state(	×
115	&self,	×
116	table_id: TableId,	×
117	) -> EtlResult<TableReplicationPhase> {	×
118	self.inner.rollback_table_replication_state(table_id).await	×
119	}	×
120
121	async fn get_table_mapping(&self, source_table_id: &TableId) -> EtlResult<Option<String>> {	×
122	self.inner.get_table_mapping(source_table_id).await	×
123	}	×
124
NEW 125	async fn get_table_mappings(&self) -> EtlResult<TableMappings> {	×
126	self.inner.get_table_mappings().await	×
127	}	×
128
129	async fn load_table_mappings(&self) -> EtlResult<usize> {	×
130	self.inner.load_table_mappings().await	×
131	}	×
132
133	async fn store_table_mapping(	×
134	&self,	×
135	source_table_id: TableId,	×
136	destination_table_id: String,	×
137	) -> EtlResult<()> {	×
138	self.inner	×
139	.store_table_mapping(source_table_id, destination_table_id)	×
140	.await	×
141	}	×
142	}
143
144	impl<S> SchemaStore for ErrorReportingStateStore<S>
145	where
146	S: SchemaStore + Send + Sync,
147	{
148	async fn get_table_schema(&self, table_id: &TableId) -> EtlResult<Option<Arc<TableSchema>>> {	×
149	self.inner.get_table_schema(table_id).await	×
150	}	×
151
152	async fn get_table_schemas(&self) -> EtlResult<Vec<Arc<TableSchema>>> {	×
153	self.inner.get_table_schemas().await	×
154	}	×
155
156	async fn load_table_schemas(&self) -> EtlResult<usize> {	×
157	self.inner.load_table_schemas().await	×
158	}	×
159
160	async fn store_table_schema(&self, table_schema: TableSchema) -> EtlResult<Arc<TableSchema>> {	×
161	self.inner.store_table_schema(table_schema).await	×
162	}	×
163	}
164
165	impl<S> CleanupStore for ErrorReportingStateStore<S>
166	where
167	S: CleanupStore + Send + Sync,
168	{
169	async fn cleanup_table_state(&self, table_id: TableId) -> EtlResult<()> {	×
170	self.inner.cleanup_table_state(table_id).await	×
171	}	×
172	}

supabase / etl / 24391673810

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous