lenso_service/
extraction_backfill.rs

1use crate::{
2    ExtractionPlan, ExtractionRun, ExtractionRunStatus, extraction_input_digest,
3    extraction_plan_integrity_is_valid, extraction_run_integrity_is_valid,
4};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use std::collections::{BTreeMap, BTreeSet};
9use std::fmt;
10
11pub const EXTRACTION_BACKFILL_PROTOCOL: &str = "lenso.extraction-backfill.v1";
12
13#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
14#[serde(rename_all = "snake_case", tag = "kind")]
15pub enum ExtractionBackfillBoundary {
16    TrustworthyCursor {
17        cursor: String,
18        source_high_water_mark: String,
19    },
20    BoundedWritePause {
21        source_high_water_mark: String,
22    },
23    Missing,
24}
25
26impl ExtractionBackfillBoundary {
27    fn high_water_mark(&self) -> Option<&str> {
28        match self {
29            Self::TrustworthyCursor {
30                source_high_water_mark,
31                ..
32            }
33            | Self::BoundedWritePause {
34                source_high_water_mark,
35            } => Some(source_high_water_mark),
36            Self::Missing => None,
37        }
38    }
39}
40
41#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
42#[serde(rename_all = "camelCase")]
43pub struct ExtractionBackfillRecord {
44    pub stable_id: String,
45    pub record_digest: String,
46    pub value: Value,
47}
48
49impl ExtractionBackfillRecord {
50    #[must_use]
51    pub fn new(stable_id: impl Into<String>, value: Value) -> Self {
52        let stable_id = stable_id.into();
53        let record_digest = digest(&(&stable_id, &value));
54        Self {
55            stable_id,
56            record_digest,
57            value,
58        }
59    }
60
61    fn integrity_is_valid(&self) -> bool {
62        self.record_digest == digest(&(&self.stable_id, &self.value))
63    }
64}
65
66#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
67#[serde(rename_all = "camelCase")]
68pub struct ExtractionBackfillScope {
69    pub plan_id: String,
70    pub plan_digest: String,
71    pub source_owner_id: String,
72    pub source_store_id: String,
73    pub destination_store_id: String,
74    pub table_mappings: Vec<String>,
75}
76
77#[derive(
78    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema,
79)]
80#[serde(rename_all = "snake_case")]
81pub enum ExtractionBackfillStatus {
82    Planned,
83    InProgress,
84    Succeeded,
85    Blocked,
86}
87
88#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
89#[serde(rename_all = "camelCase")]
90pub struct ExtractionBackfillRequest {
91    pub batch_id: String,
92    #[serde(default, skip_serializing_if = "Option::is_none")]
93    pub expected_destination_checkpoint: Option<String>,
94    pub records: Vec<ExtractionBackfillRecord>,
95    pub final_batch: bool,
96}
97
98impl ExtractionBackfillRequest {
99    #[must_use]
100    pub fn new(
101        batch_id: impl Into<String>,
102        expected_destination_checkpoint: Option<String>,
103        records: Vec<ExtractionBackfillRecord>,
104    ) -> Self {
105        Self {
106            batch_id: batch_id.into(),
107            expected_destination_checkpoint,
108            records,
109            final_batch: false,
110        }
111    }
112
113    #[must_use]
114    pub fn final_batch(mut self) -> Self {
115        self.final_batch = true;
116        self
117    }
118}
119
120#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
121#[serde(rename_all = "camelCase")]
122pub struct ExtractionBackfillBatchReceipt {
123    pub batch_id: String,
124    pub batch_digest: String,
125    pub previous_destination_checkpoint: Option<String>,
126    pub destination_checkpoint: String,
127    pub first_stable_id: Option<String>,
128    pub last_stable_id: Option<String>,
129    pub copied_count: u64,
130    pub duplicate_count: u64,
131    pub source_high_water_mark: String,
132    pub source_authority_unchanged: bool,
133    pub candidate_authoritative: bool,
134}
135
136#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
137#[serde(rename_all = "camelCase")]
138pub struct ExtractionBackfillProgress {
139    pub copied_count: u64,
140    pub remaining_lag: u64,
141    pub source_high_water_mark: String,
142    #[serde(default, skip_serializing_if = "Option::is_none")]
143    pub destination_checkpoint: Option<String>,
144    #[serde(default, skip_serializing_if = "Option::is_none")]
145    pub next_after_stable_id: Option<String>,
146}
147
148#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
149#[serde(rename_all = "camelCase")]
150pub struct ExtractionBackfillEvidence {
151    pub kind: String,
152    pub subject: String,
153    pub digest: String,
154    pub detail: String,
155}
156
157#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
158#[serde(rename_all = "camelCase")]
159pub struct ExtractionBackfillEffects {
160    pub reads_plan_scoped_source_data: bool,
161    pub copies_destination_data: bool,
162    pub mutates_source_data: bool,
163    pub changes_authority: bool,
164    pub emits_business_effects: bool,
165}
166
167#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
168#[serde(rename_all = "camelCase")]
169pub struct ExtractionBackfillRun {
170    pub protocol: String,
171    pub run_id: String,
172    pub run_digest: String,
173    pub revision: u64,
174    pub status: ExtractionBackfillStatus,
175    pub scope: ExtractionBackfillScope,
176    pub boundary: ExtractionBackfillBoundary,
177    pub progress: ExtractionBackfillProgress,
178    #[serde(default)]
179    pub destination_records: Vec<ExtractionBackfillRecord>,
180    #[serde(default)]
181    pub receipts: Vec<ExtractionBackfillBatchReceipt>,
182    #[serde(default)]
183    pub evidence: Vec<ExtractionBackfillEvidence>,
184    #[serde(default)]
185    pub next_actions: Vec<String>,
186    pub linked_authority_remains_authoritative: bool,
187    pub candidate_authoritative: bool,
188    pub effects: ExtractionBackfillEffects,
189}
190
191#[derive(
192    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema,
193)]
194#[serde(rename_all = "snake_case")]
195pub enum ExtractionBackfillErrorCode {
196    PlanInvalid,
197    DestinationExpansionIncomplete,
198    BackfillCursorMissing,
199    BackfillBatchUnordered,
200    BackfillCheckpointStale,
201    BackfillBatchChanged,
202    BackfillRecordChanged,
203    BackfillRunInvalid,
204    BackfillPersistenceFailed,
205}
206
207impl ExtractionBackfillErrorCode {
208    #[must_use]
209    pub const fn as_str(self) -> &'static str {
210        match self {
211            Self::PlanInvalid => "plan_invalid",
212            Self::DestinationExpansionIncomplete => "destination_expansion_incomplete",
213            Self::BackfillCursorMissing => "backfill_cursor_missing",
214            Self::BackfillBatchUnordered => "backfill_batch_unordered",
215            Self::BackfillCheckpointStale => "backfill_checkpoint_stale",
216            Self::BackfillBatchChanged => "backfill_batch_changed",
217            Self::BackfillRecordChanged => "backfill_record_changed",
218            Self::BackfillRunInvalid => "backfill_run_invalid",
219            Self::BackfillPersistenceFailed => "backfill_persistence_failed",
220        }
221    }
222}
223
224#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
225#[serde(rename_all = "camelCase")]
226pub struct ExtractionBackfillError {
227    pub code: ExtractionBackfillErrorCode,
228    pub message: String,
229    pub next_actions: Vec<String>,
230}
231
232impl fmt::Display for ExtractionBackfillError {
233    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
234        formatter.write_str(&self.message)
235    }
236}
237
238impl std::error::Error for ExtractionBackfillError {}
239
240pub fn start_extraction_backfill(
241    plan: &ExtractionPlan,
242    expansion: &ExtractionRun,
243    boundary: ExtractionBackfillBoundary,
244) -> Result<ExtractionBackfillRun, ExtractionBackfillError> {
245    if !extraction_plan_integrity_is_valid(plan) {
246        return Err(error(
247            ExtractionBackfillErrorCode::PlanInvalid,
248            "Extraction Plan integrity validation failed before backfill.",
249            "Regenerate and review the Extraction Plan.",
250        ));
251    }
252    if !extraction_run_integrity_is_valid(expansion)
253        || expansion.current_phase.status != ExtractionRunStatus::Succeeded
254        || expansion.plan.plan_id != plan.plan_id
255        || expansion.plan.plan_digest != plan.plan_digest
256    {
257        return Err(error(
258            ExtractionBackfillErrorCode::DestinationExpansionIncomplete,
259            "Destination expansion evidence is incomplete or belongs to another plan.",
260            "Finish destination expansion for this exact Extraction Plan.",
261        ));
262    }
263    let Some(source_high_water_mark) = boundary.high_water_mark().map(str::to_owned) else {
264        return Err(error(
265            ExtractionBackfillErrorCode::BackfillCursorMissing,
266            "Online backfill requires a trustworthy cursor or a bounded write pause.",
267            "Declare a trustworthy extraction cursor or enter the protected write-pause phase.",
268        ));
269    };
270    if source_high_water_mark.is_empty() {
271        return Err(error(
272            ExtractionBackfillErrorCode::BackfillCursorMissing,
273            "The source high-water mark must not be empty.",
274            "Capture a durable source high-water mark before copying data.",
275        ));
276    }
277    let mut table_mappings = plan
278        .data_mapping
279        .tables
280        .iter()
281        .map(|mapping| format!("{}->{}", mapping.source_table, mapping.destination_table))
282        .collect::<Vec<_>>();
283    table_mappings.sort();
284    let scope = ExtractionBackfillScope {
285        plan_id: plan.plan_id.clone(),
286        plan_digest: plan.plan_digest.clone(),
287        source_owner_id: plan.expected_authority.owner_id.clone(),
288        source_store_id: format!("linked:{}", plan.expected_authority.owner_id),
289        destination_store_id: plan.proposed_service.store.store_id.clone(),
290        table_mappings,
291    };
292    let identity = digest(&(&scope, &boundary));
293    let mut run = ExtractionBackfillRun {
294        protocol: EXTRACTION_BACKFILL_PROTOCOL.to_owned(),
295        run_id: format!("extraction-backfill:{identity}"),
296        run_digest: String::new(),
297        revision: 1,
298        status: ExtractionBackfillStatus::Planned,
299        scope,
300        boundary,
301        progress: ExtractionBackfillProgress {
302            copied_count: 0,
303            remaining_lag: 1,
304            source_high_water_mark: source_high_water_mark.clone(),
305            destination_checkpoint: None,
306            next_after_stable_id: None,
307        },
308        destination_records: Vec::new(),
309        receipts: Vec::new(),
310        evidence: vec![ExtractionBackfillEvidence {
311            kind: "source_boundary".to_owned(),
312            subject: source_high_water_mark.clone(),
313            digest: extraction_input_digest(source_high_water_mark.as_bytes()),
314            detail: "Backfill is bounded by plan-scoped source evidence.".to_owned(),
315        }],
316        next_actions: vec!["Copy the next deterministic Postgres batch and persist its receipt atomically with the destination checkpoint.".to_owned()],
317        linked_authority_remains_authoritative: true,
318        candidate_authoritative: false,
319        effects: ExtractionBackfillEffects::default(),
320    };
321    refresh_digest(&mut run);
322    Ok(run)
323}
324
325pub fn apply_extraction_backfill_batch(
326    mut run: ExtractionBackfillRun,
327    request: ExtractionBackfillRequest,
328) -> Result<ExtractionBackfillRun, ExtractionBackfillError> {
329    if !extraction_backfill_integrity_is_valid(&run) {
330        return Err(error(
331            ExtractionBackfillErrorCode::BackfillRunInvalid,
332            "Backfill Run integrity validation failed.",
333            "Resume from the last integrity-valid durable revision.",
334        ));
335    }
336    let batch_digest = digest(&request);
337    if let Some(receipt) = run
338        .receipts
339        .iter()
340        .find(|receipt| receipt.batch_id == request.batch_id)
341    {
342        if receipt.batch_digest == batch_digest {
343            return Ok(run);
344        }
345        return Err(error(
346            ExtractionBackfillErrorCode::BackfillBatchChanged,
347            "A committed batch id was reused with different contents.",
348            "Resume with the original batch or allocate the next ordered batch id.",
349        ));
350    }
351    if request.expected_destination_checkpoint != run.progress.destination_checkpoint {
352        return Err(error(
353            ExtractionBackfillErrorCode::BackfillCheckpointStale,
354            "The requested destination checkpoint is stale.",
355            "Reload the durable Backfill Run and resume from its current checkpoint.",
356        ));
357    }
358    let ordered_ids = request
359        .records
360        .iter()
361        .map(|record| record.stable_id.as_str())
362        .collect::<Vec<_>>();
363    if ordered_ids
364        .windows(2)
365        .any(|pair| !stable_id_is_strictly_before(pair[0], pair[1]))
366        || request
367            .records
368            .iter()
369            .any(|record| !record.integrity_is_valid())
370    {
371        return Err(error(
372            ExtractionBackfillErrorCode::BackfillBatchUnordered,
373            "Backfill records must have unique stable identities in deterministic ascending order.",
374            "Sort the source query by the declared stable identity and rebuild the batch.",
375        ));
376    }
377    if run
378        .progress
379        .next_after_stable_id
380        .as_deref()
381        .is_some_and(|last| {
382            request.records.first().is_some_and(|record| {
383                !stable_id_is_strictly_before(last, record.stable_id.as_str())
384            })
385        })
386    {
387        return Err(error(
388            ExtractionBackfillErrorCode::BackfillBatchUnordered,
389            "The next batch does not advance beyond the durable stable-identity checkpoint.",
390            "Resume the source query strictly after nextAfterStableId.",
391        ));
392    }
393    let mut destination = run
394        .destination_records
395        .iter()
396        .cloned()
397        .map(|record| (record.stable_id.clone(), record))
398        .collect::<BTreeMap<_, _>>();
399    let mut copied_count = 0_u64;
400    let mut duplicate_count = 0_u64;
401    for record in &request.records {
402        match destination.get(&record.stable_id) {
403            Some(existing) if existing == record => duplicate_count += 1,
404            Some(_) => {
405                return Err(error(
406                    ExtractionBackfillErrorCode::BackfillRecordChanged,
407                    "A stable source identity changed after it was checkpointed.",
408                    "Capture a fresh source boundary and regenerate the affected batch.",
409                ));
410            }
411            None => {
412                destination.insert(record.stable_id.clone(), record.clone());
413                copied_count += 1;
414            }
415        }
416    }
417    let destination_checkpoint = format!(
418        "backfill-checkpoint:{}",
419        digest(&(
420            run.run_id.as_str(),
421            run.progress.destination_checkpoint.as_deref(),
422            request.batch_id.as_str(),
423            batch_digest.as_str(),
424        ))
425    );
426    let first_stable_id = request
427        .records
428        .first()
429        .map(|record| record.stable_id.clone());
430    let last_stable_id = request
431        .records
432        .last()
433        .map(|record| record.stable_id.clone());
434    let receipt = ExtractionBackfillBatchReceipt {
435        batch_id: request.batch_id,
436        batch_digest,
437        previous_destination_checkpoint: run.progress.destination_checkpoint.clone(),
438        destination_checkpoint: destination_checkpoint.clone(),
439        first_stable_id,
440        last_stable_id: last_stable_id.clone(),
441        copied_count,
442        duplicate_count,
443        source_high_water_mark: run.progress.source_high_water_mark.clone(),
444        source_authority_unchanged: true,
445        candidate_authoritative: false,
446    };
447    run.destination_records = destination.into_values().collect();
448    run.receipts.push(receipt.clone());
449    run.progress.copied_count += copied_count;
450    run.progress.destination_checkpoint = Some(destination_checkpoint);
451    run.progress.next_after_stable_id = last_stable_id;
452    run.progress.remaining_lag = u64::from(!request.final_batch);
453    run.status = if request.final_batch {
454        ExtractionBackfillStatus::Succeeded
455    } else {
456        ExtractionBackfillStatus::InProgress
457    };
458    run.effects.reads_plan_scoped_source_data = true;
459    run.effects.copies_destination_data |= copied_count > 0;
460    run.evidence.push(ExtractionBackfillEvidence {
461        kind: "durable_batch_receipt".to_owned(),
462        subject: receipt.batch_id.clone(),
463        digest: receipt.batch_digest.clone(),
464        detail: format!(
465            "Copied {} records and observed {} already-checkpointed records.",
466            receipt.copied_count, receipt.duplicate_count
467        ),
468    });
469    run.next_actions = if request.final_batch {
470        vec!["Reconcile the candidate Store against this exact source high-water mark and destination checkpoint.".to_owned()]
471    } else {
472        vec!["Persist this receipt, then request the next ordered source batch after nextAfterStableId.".to_owned()]
473    };
474    run.revision += 1;
475    refresh_digest(&mut run);
476    Ok(run)
477}
478
479/// Atomically persists destination records, the batch receipt, and the next
480/// checkpoint in PostgreSQL. The run row is locked and compared by digest so a
481/// restarted or concurrent orchestrator cannot overwrite newer progress.
482pub async fn apply_postgres_extraction_backfill_batch(
483    pool: &sqlx::PgPool,
484    run: ExtractionBackfillRun,
485    request: ExtractionBackfillRequest,
486) -> Result<ExtractionBackfillRun, ExtractionBackfillError> {
487    let mut transaction = pool.begin().await.map_err(persistence_error)?;
488    sqlx::query("create schema if not exists lenso_extraction")
489        .execute(&mut *transaction)
490        .await
491        .map_err(persistence_error)?;
492    sqlx::query(
493        r#"
494        create table if not exists lenso_extraction.backfill_runs (
495            run_id text primary key,
496            revision bigint not null,
497            run_digest text not null,
498            run_json jsonb not null,
499            updated_at timestamptz not null default now()
500        )
501        "#,
502    )
503    .execute(&mut *transaction)
504    .await
505    .map_err(persistence_error)?;
506    sqlx::query(
507        r#"
508        create table if not exists lenso_extraction.backfill_records (
509            run_id text not null references lenso_extraction.backfill_runs(run_id),
510            stable_id text not null,
511            record_digest text not null,
512            record_json jsonb not null,
513            primary key (run_id, stable_id)
514        )
515        "#,
516    )
517    .execute(&mut *transaction)
518    .await
519    .map_err(persistence_error)?;
520    sqlx::query("select pg_advisory_xact_lock(hashtext($1))")
521        .bind(&run.run_id)
522        .execute(&mut *transaction)
523        .await
524        .map_err(persistence_error)?;
525
526    let stored = sqlx::query_as::<_, (i64, String, serde_json::Value)>(
527        "select revision, run_digest, run_json from lenso_extraction.backfill_runs where run_id = $1 for update",
528    )
529    .bind(&run.run_id)
530    .fetch_optional(&mut *transaction)
531    .await
532    .map_err(persistence_error)?;
533    let durable_run = if let Some((_, stored_digest, stored_json)) = stored {
534        let stored_run: ExtractionBackfillRun =
535            serde_json::from_value(stored_json).map_err(|source| {
536                error(
537                    ExtractionBackfillErrorCode::BackfillPersistenceFailed,
538                    format!("Stored Backfill Run is unreadable: {source}"),
539                    "Repair or restore the last integrity-valid durable Run.",
540                )
541            })?;
542        if stored_digest != run.run_digest {
543            let replayed = apply_extraction_backfill_batch(run.clone(), request.clone())?;
544            let expected_receipt = replayed
545                .receipts
546                .iter()
547                .find(|receipt| receipt.batch_id == request.batch_id);
548            let stored_receipt = stored_run
549                .receipts
550                .iter()
551                .find(|receipt| receipt.batch_id == request.batch_id);
552            if expected_receipt == stored_receipt && stored_receipt.is_some() {
553                transaction.commit().await.map_err(persistence_error)?;
554                return Ok(stored_run);
555            }
556            return Err(error(
557                ExtractionBackfillErrorCode::BackfillCheckpointStale,
558                "A newer durable PostgreSQL checkpoint already exists for this Backfill Run.",
559                "Reload the durable Run and resume from its current checkpoint.",
560            ));
561        }
562        stored_run
563    } else {
564        sqlx::query(
565            "insert into lenso_extraction.backfill_runs (run_id, revision, run_digest, run_json) values ($1, $2, $3, $4)",
566        )
567        .bind(&run.run_id)
568        .bind(i64::try_from(run.revision).unwrap_or(i64::MAX))
569        .bind(&run.run_digest)
570        .bind(serde_json::to_value(&run).map_err(|source| {
571            error(
572                ExtractionBackfillErrorCode::BackfillPersistenceFailed,
573                format!("Backfill Run could not serialize: {source}"),
574                "Persist an integrity-valid Backfill Run.",
575            )
576        })?)
577        .execute(&mut *transaction)
578        .await
579        .map_err(persistence_error)?;
580        run
581    };
582    let previous_revision = durable_run.revision;
583    let next = apply_extraction_backfill_batch(durable_run, request)?;
584    for record in &next.destination_records {
585        let persisted = sqlx::query(
586            r#"
587            insert into lenso_extraction.backfill_records (run_id, stable_id, record_digest, record_json)
588            values ($1, $2, $3, $4)
589            on conflict (run_id, stable_id) do update
590            set record_digest = excluded.record_digest, record_json = excluded.record_json
591            where lenso_extraction.backfill_records.record_digest = excluded.record_digest
592            "#,
593        )
594        .bind(&next.run_id)
595        .bind(&record.stable_id)
596        .bind(&record.record_digest)
597        .bind(&record.value)
598        .execute(&mut *transaction)
599        .await
600        .map_err(persistence_error)?;
601        if persisted.rows_affected() != 1 {
602            return Err(error(
603                ExtractionBackfillErrorCode::BackfillRecordChanged,
604                format!(
605                    "Durable candidate record {} differs from the checkpointed Backfill Run.",
606                    record.stable_id
607                ),
608                "Reconcile the durable record ledger before advancing the checkpoint.",
609            ));
610        }
611    }
612    let updated = sqlx::query(
613        r#"
614        update lenso_extraction.backfill_runs
615        set revision = $2, run_digest = $3, run_json = $4, updated_at = now()
616        where run_id = $1 and revision = $5
617        "#,
618    )
619    .bind(&next.run_id)
620    .bind(i64::try_from(next.revision).unwrap_or(i64::MAX))
621    .bind(&next.run_digest)
622    .bind(serde_json::to_value(&next).map_err(|source| {
623        error(
624            ExtractionBackfillErrorCode::BackfillPersistenceFailed,
625            format!("Backfill Run could not serialize: {source}"),
626            "Persist an integrity-valid Backfill Run.",
627        )
628    })?)
629    .bind(i64::try_from(previous_revision).unwrap_or(i64::MAX))
630    .execute(&mut *transaction)
631    .await
632    .map_err(persistence_error)?;
633    if updated.rows_affected() != 1 {
634        return Err(error(
635            ExtractionBackfillErrorCode::BackfillCheckpointStale,
636            "The durable PostgreSQL checkpoint changed during this batch.",
637            "Reload the durable Run and retry from its current checkpoint.",
638        ));
639    }
640    transaction.commit().await.map_err(persistence_error)?;
641    Ok(next)
642}
643
644/// Reload the last transactionally committed run after process restart or a
645/// lost client response.
646pub async fn load_postgres_extraction_backfill(
647    pool: &sqlx::PgPool,
648    run_id: &str,
649) -> Result<Option<ExtractionBackfillRun>, ExtractionBackfillError> {
650    let exists = sqlx::query_scalar::<_, Option<String>>(
651        "select to_regclass('lenso_extraction.backfill_runs')::text",
652    )
653    .fetch_one(pool)
654    .await
655    .map_err(persistence_error)?
656    .is_some();
657    if !exists {
658        return Ok(None);
659    }
660    let value = sqlx::query_scalar::<_, serde_json::Value>(
661        "select run_json from lenso_extraction.backfill_runs where run_id = $1",
662    )
663    .bind(run_id)
664    .fetch_optional(pool)
665    .await
666    .map_err(persistence_error)?;
667    value
668        .map(|value| {
669            serde_json::from_value(value).map_err(|source| {
670                error(
671                    ExtractionBackfillErrorCode::BackfillPersistenceFailed,
672                    format!("Stored Backfill Run is unreadable: {source}"),
673                    "Repair or restore the last integrity-valid durable Run.",
674                )
675            })
676        })
677        .transpose()
678}
679
680/// Read one plan-scoped source batch and copy it into the planned candidate
681/// Postgres table before atomically advancing the durable checkpoint.
682pub async fn copy_postgres_extraction_service_data_batch(
683    source_pool: &sqlx::PgPool,
684    destination_pool: &sqlx::PgPool,
685    plan: &ExtractionPlan,
686    run: ExtractionBackfillRun,
687    batch_id: impl Into<String>,
688    limit: i64,
689) -> Result<ExtractionBackfillRun, ExtractionBackfillError> {
690    if !extraction_plan_integrity_is_valid(plan)
691        || run.scope.plan_id != plan.plan_id
692        || run.scope.plan_digest != plan.plan_digest
693    {
694        return Err(error(
695            ExtractionBackfillErrorCode::BackfillRunInvalid,
696            "Backfill Run does not belong to the supplied Extraction Plan.",
697            "Load the exact plan-scoped Run before reading Service Data.",
698        ));
699    }
700    if plan.data_mapping.tables.len() != 1 {
701        return Err(error(
702            ExtractionBackfillErrorCode::BackfillRunInvalid,
703            "A Postgres Backfill Run must be scoped to exactly one table mapping.",
704            "Create one durable plan-scoped Backfill Run per table mapping.",
705        ));
706    }
707    let mapping = plan.data_mapping.tables.first().ok_or_else(|| {
708        error(
709            ExtractionBackfillErrorCode::BackfillRunInvalid,
710            "Extraction Plan has no Postgres table mapping.",
711            "Regenerate the plan with an owned source and destination table.",
712        )
713    })?;
714    let cursor = mapping
715        .cursors
716        .iter()
717        .find(|cursor| cursor.trustworthy)
718        .ok_or_else(|| {
719            error(
720                ExtractionBackfillErrorCode::BackfillCursorMissing,
721                "Extraction Plan has no trustworthy Postgres cursor.",
722                "Enter the bounded write-pause phase or regenerate cursor evidence.",
723            )
724        })?;
725    let source_table = quoted_relation(&mapping.source_table)?;
726    let destination_table = quoted_relation(&mapping.destination_table)?;
727    let cursor_column = quoted_identifier(&cursor.column)?;
728    let cursor_name = cursor.column.as_str();
729    let after_cursor = format!(
730        "(jsonb_populate_record(null::{source_table}, jsonb_build_object('{cursor_name}', $1::text))).{cursor_column}"
731    );
732    let high_water_cursor = format!(
733        "(jsonb_populate_record(null::{source_table}, jsonb_build_object('{cursor_name}', $2::text))).{cursor_column}"
734    );
735    let after = run
736        .progress
737        .next_after_stable_id
738        .clone()
739        .unwrap_or_default();
740    let rows = sqlx::query_as::<_, (String, serde_json::Value)>(sqlx::AssertSqlSafe(format!(
741        "select {cursor_column}::text, to_jsonb(source_row) from {source_table} source_row where ($1 = '' or {cursor_column} > {after_cursor}) and {cursor_column} <= {high_water_cursor} order by {cursor_column} limit $3"
742    )))
743    .bind(&after)
744    .bind(&run.progress.source_high_water_mark)
745    .bind(limit.max(1))
746    .fetch_all(source_pool)
747    .await
748    .map_err(persistence_error)?;
749    let mut transaction = destination_pool.begin().await.map_err(persistence_error)?;
750    let mut records = Vec::with_capacity(rows.len());
751    for (stable_id, value) in rows {
752        let existing = sqlx::query_scalar::<_, serde_json::Value>(sqlx::AssertSqlSafe(format!(
753            "select to_jsonb(destination_row) from {destination_table} destination_row where {cursor_column}::text = $1"
754        )))
755        .bind(&stable_id)
756        .fetch_optional(&mut *transaction)
757        .await
758        .map_err(persistence_error)?;
759        if let Some(existing) = existing {
760            if existing != value {
761                return Err(error(
762                    ExtractionBackfillErrorCode::BackfillRecordChanged,
763                    format!("Candidate record {stable_id} differs from the plan-scoped source."),
764                    "Reconcile the conflicting candidate record before resuming.",
765                ));
766            }
767        } else {
768            sqlx::query(sqlx::AssertSqlSafe(format!(
769                "insert into {destination_table} select * from jsonb_populate_record(null::{destination_table}, $1)"
770            )))
771            .bind(&value)
772            .execute(&mut *transaction)
773            .await
774            .map_err(persistence_error)?;
775        }
776        records.push(ExtractionBackfillRecord::new(stable_id, value));
777    }
778    transaction.commit().await.map_err(persistence_error)?;
779    let final_batch = records
780        .last()
781        .is_none_or(|record| record.stable_id == run.progress.source_high_water_mark)
782        || i64::try_from(records.len()).unwrap_or(i64::MAX) < limit.max(1);
783    let mut request = ExtractionBackfillRequest::new(
784        batch_id,
785        run.progress.destination_checkpoint.clone(),
786        records,
787    );
788    request.final_batch = final_batch;
789    apply_postgres_extraction_backfill_batch(destination_pool, run, request).await
790}
791
792fn quoted_relation(value: &str) -> Result<String, ExtractionBackfillError> {
793    value
794        .split('.')
795        .map(quoted_identifier)
796        .collect::<Result<Vec<_>, _>>()
797        .map(|parts| parts.join("."))
798}
799
800fn quoted_identifier(value: &str) -> Result<String, ExtractionBackfillError> {
801    if value.is_empty()
802        || !value
803            .chars()
804            .all(|character| character.is_ascii_alphanumeric() || character == '_')
805    {
806        return Err(error(
807            ExtractionBackfillErrorCode::BackfillRunInvalid,
808            format!("Unsafe Postgres identifier `{value}` in Extraction Plan."),
809            "Regenerate the plan from validated schema ownership evidence.",
810        ));
811    }
812    Ok(format!("\"{value}\""))
813}
814
815fn persistence_error(source: sqlx::Error) -> ExtractionBackfillError {
816    error(
817        ExtractionBackfillErrorCode::BackfillPersistenceFailed,
818        format!("PostgreSQL backfill persistence failed: {source}"),
819        "Restore PostgreSQL availability and resume from the last durable checkpoint.",
820    )
821}
822
823#[must_use]
824pub fn extraction_backfill_integrity_is_valid(run: &ExtractionBackfillRun) -> bool {
825    if run.protocol != EXTRACTION_BACKFILL_PROTOCOL
826        || !run.linked_authority_remains_authoritative
827        || run.candidate_authoritative
828        || run.effects.mutates_source_data
829        || run.effects.changes_authority
830        || run.effects.emits_business_effects
831        || run.progress.source_high_water_mark != run.boundary.high_water_mark().unwrap_or_default()
832        || run
833            .destination_records
834            .iter()
835            .any(|record| !record.integrity_is_valid())
836        || run
837            .destination_records
838            .windows(2)
839            .any(|pair| !stable_id_is_strictly_before(&pair[0].stable_id, &pair[1].stable_id))
840    {
841        return false;
842    }
843    let receipt_ids = run
844        .receipts
845        .iter()
846        .map(|receipt| receipt.batch_id.as_str())
847        .collect::<BTreeSet<_>>();
848    receipt_ids.len() == run.receipts.len() && run.run_digest == run_digest(run)
849}
850
851fn stable_id_is_strictly_before(left: &str, right: &str) -> bool {
852    match (left.parse::<i128>(), right.parse::<i128>()) {
853        (Ok(left), Ok(right)) => left < right,
854        _ => left < right,
855    }
856}
857
858fn refresh_digest(run: &mut ExtractionBackfillRun) {
859    run.run_digest = run_digest(run);
860}
861
862fn run_digest(run: &ExtractionBackfillRun) -> String {
863    let mut value = run.clone();
864    value.run_digest.clear();
865    digest(&value)
866}
867
868fn digest(value: &impl Serialize) -> String {
869    let bytes = serde_json::to_vec(value).expect("Extraction backfill values must serialize");
870    extraction_input_digest(&bytes)
871}
872
873fn error(
874    code: ExtractionBackfillErrorCode,
875    message: impl Into<String>,
876    next_action: impl Into<String>,
877) -> ExtractionBackfillError {
878    ExtractionBackfillError {
879        code,
880        message: message.into(),
881        next_actions: vec![next_action.into()],
882    }
883}
lenso_service/extraction_backfill.rs

lenso_service/
extraction_backfill.rs
