@@ -281,9 +281,9 @@ pub struct PostgresPhysicalReplicaStatus {
281281#[ serde( default , skip_serializing_if = "Option::is_none" ) ]
282282 pub schema_migration_job : Option < String > ,
283283
284- /// Phase of schema migration: pending, active, complete, failed
284+ /// Phase of schema migration. See [`SchemaMigrationPhase`].
285285#[ serde( default , skip_serializing_if = "Option::is_none" ) ]
286- pub schema_migration_phase : Option < String > ,
286+ pub schema_migration_phase : Option < SchemaMigrationPhase > ,
287287
288288 /// Measured size of persistent schema data from the last successful migration (bytes).
289289/// Used to size the next restore PVC.
@@ -306,6 +306,106 @@ pub enum ReplicaPhase {
306306 Failed ,
307307}
308308
309+ /// Lifecycle phase of the operator's schema-migration step that runs
310+ /// during a switchover when `persistent_schemas` is configured on the
311+ /// replica. The serialized form is a flat string matching the historical
312+ /// wire format (`active` / `complete` / `partial` / `timeout-skipped` /
313+ /// `failed: <reason>`) so existing replica status objects round-trip
314+ /// unchanged.
315+ #[ derive( Debug , Clone , PartialEq , Eq ) ]
316+ pub enum SchemaMigrationPhase {
317+ /// Migration Job is running. The sweep must not delete the source
318+ /// restore while we're in this state.
319+ Active ,
320+ /// Migration Job finished cleanly; persistent schemas were carried
321+ /// across to the new restore.
322+ Complete ,
323+ /// Migration Job finished but psql logged statement errors (typical
324+ /// when dbt views reference renamed/dropped upstream columns). Some
325+ /// persistent_schemas objects may need regenerating upstream.
326+ Partial ,
327+ /// Migration exceeded the per-cycle wall-clock budget (20% of the
328+ /// cron interval). The operator dropped the persistent_schemas on
329+ /// the new restore and proceeded to switchover anyway — a usable
330+ /// replica beats carrying the schema through indefinitely. The next
331+ /// cycle re-attempts migration if the schemas have regenerated.
332+ TimeoutSkipped ,
333+ /// Migration Job failed. The old restore stays Active; the new
334+ /// restore stays in Switching. The wrapped string is the reason
335+ /// surfaced from the Job's callback body (or "no callback received").
336+ Failed ( String ) ,
337+ }
338+
339+ impl SchemaMigrationPhase {
340+ /// True for every phase except [`Self::Active`]. Used by the sweep
341+ /// gate: as long as the migration isn't currently running, deleting
342+ /// the previous Active restore is safe (nothing depends on it being
343+ /// around). Coded as "not Active" rather than enumerating terminal
344+ /// variants so adding a new variant doesn't risk silently
345+ /// reintroducing the deadlock that originally motivated this enum.
346+ pub fn is_settled ( & self ) -> bool {
347+ !matches ! ( self , Self :: Active )
348+ }
349+ }
350+
351+ impl std:: fmt:: Display for SchemaMigrationPhase {
352+ fn fmt ( & self , f : & mut std:: fmt:: Formatter < ' _ > ) -> std:: fmt:: Result {
353+ match self {
354+ Self :: Active => f. write_str ( "active" ) ,
355+ Self :: Complete => f. write_str ( "complete" ) ,
356+ Self :: Partial => f. write_str ( "partial" ) ,
357+ Self :: TimeoutSkipped => f. write_str ( "timeout-skipped" ) ,
358+ Self :: Failed ( reason) => write ! ( f, "failed: {reason}" ) ,
359+ }
360+ }
361+ }
362+
363+ impl std:: str:: FromStr for SchemaMigrationPhase {
364+ type Err = String ;
365+
366+ fn from_str ( s : & str ) -> std:: result:: Result < Self , Self :: Err > {
367+ match s {
368+ "active" => Ok ( Self :: Active ) ,
369+ "complete" => Ok ( Self :: Complete ) ,
370+ "partial" => Ok ( Self :: Partial ) ,
371+ "timeout-skipped" => Ok ( Self :: TimeoutSkipped ) ,
372+ other => {
373+ if let Some ( reason) = other. strip_prefix ( "failed:" ) {
374+ Ok ( Self :: Failed ( reason. trim ( ) . to_string ( ) ) )
375+ } else {
376+ Err ( format ! ( "unknown schema migration phase: {other:?}" ) )
377+ }
378+ }
379+ }
380+ }
381+ }
382+
383+ impl Serialize for SchemaMigrationPhase {
384+ fn serialize < S : serde:: Serializer > ( & self , s : S ) -> Result < S :: Ok , S :: Error > {
385+ s. serialize_str ( & self . to_string ( ) )
386+ }
387+ }
388+
389+ impl < ' de > Deserialize < ' de > for SchemaMigrationPhase {
390+ fn deserialize < D : serde:: Deserializer < ' de > > ( d : D ) -> Result < Self , D :: Error > {
391+ let s = String :: deserialize ( d) ?;
392+ s. parse ( ) . map_err ( serde:: de:: Error :: custom)
393+ }
394+ }
395+
396+ impl JsonSchema for SchemaMigrationPhase {
397+ fn schema_name ( ) -> Cow < ' static , str > {
398+ "SchemaMigrationPhase" . into ( )
399+ }
400+
401+ fn json_schema ( _: & mut SchemaGenerator ) -> Schema {
402+ json_schema ! ( {
403+ "type" : "string" ,
404+ "description" : "Schema migration phase: 'active' (Job running), 'complete' (succeeded), 'partial' (succeeded with statement errors), 'timeout-skipped' (budget exceeded; persistent schemas dropped and switchover proceeded), or 'failed: <reason>' (Job failed)." ,
405+ } )
406+ }
407+ }
408+
309409#[ derive( Debug , Clone , Deserialize , Serialize , JsonSchema ) ]
310410#[ serde( rename_all = "camelCase" ) ]
311411pub struct ConnectionInfo {
@@ -349,3 +449,64 @@ impl PostgresPhysicalReplica {
349449 format ! ( "{name}-creds" , name = self . name_any( ) )
350450 }
351451}
452+
453+ #[ cfg( test) ]
454+ mod tests {
455+ use super :: * ;
456+
457+ #[ test]
458+ fn schema_migration_phase_roundtrips_terminal_variants ( ) {
459+ for phase in [
460+ SchemaMigrationPhase :: Active ,
461+ SchemaMigrationPhase :: Complete ,
462+ SchemaMigrationPhase :: Partial ,
463+ SchemaMigrationPhase :: TimeoutSkipped ,
464+ ] {
465+ let s = serde_json:: to_string ( & phase) . expect ( "serialize" ) ;
466+ let back: SchemaMigrationPhase = serde_json:: from_str ( & s) . expect ( "deserialize" ) ;
467+ assert_eq ! ( phase, back, "round-trip mismatch for {phase:?}" ) ;
468+ }
469+ }
470+
471+ #[ test]
472+ fn schema_migration_phase_failed_preserves_reason ( ) {
473+ let phase = SchemaMigrationPhase :: Failed ( "connection refused" . into ( ) ) ;
474+ let s = serde_json:: to_string ( & phase) . expect ( "serialize" ) ;
475+ assert_eq ! ( s, "\" failed: connection refused\" " ) ;
476+ let back: SchemaMigrationPhase = serde_json:: from_str ( & s) . expect ( "deserialize" ) ;
477+ assert_eq ! ( phase, back) ;
478+ }
479+
480+ #[ test]
481+ fn schema_migration_phase_wire_strings_match_history ( ) {
482+ // The wire format is documented in the README and consumed by
483+ // external tooling (dashboards, alerts). These strings are part
484+ // of pgro's public contract; renaming them is a breaking change.
485+ assert_eq ! ( SchemaMigrationPhase :: Active . to_string( ) , "active" ) ;
486+ assert_eq ! ( SchemaMigrationPhase :: Complete . to_string( ) , "complete" ) ;
487+ assert_eq ! ( SchemaMigrationPhase :: Partial . to_string( ) , "partial" ) ;
488+ assert_eq ! (
489+ SchemaMigrationPhase :: TimeoutSkipped . to_string( ) ,
490+ "timeout-skipped"
491+ ) ;
492+ assert_eq ! (
493+ SchemaMigrationPhase :: Failed ( "boom" . into( ) ) . to_string( ) ,
494+ "failed: boom"
495+ ) ;
496+ }
497+
498+ #[ test]
499+ fn schema_migration_phase_rejects_unknown_string ( ) {
500+ let r: Result < SchemaMigrationPhase , _ > = "what" . parse ( ) ;
501+ assert ! ( r. is_err( ) ) ;
502+ }
503+
504+ #[ test]
505+ fn schema_migration_phase_is_settled ( ) {
506+ assert ! ( !SchemaMigrationPhase :: Active . is_settled( ) ) ;
507+ assert ! ( SchemaMigrationPhase :: Complete . is_settled( ) ) ;
508+ assert ! ( SchemaMigrationPhase :: Partial . is_settled( ) ) ;
509+ assert ! ( SchemaMigrationPhase :: TimeoutSkipped . is_settled( ) ) ;
510+ assert ! ( SchemaMigrationPhase :: Failed ( "x" . into( ) ) . is_settled( ) ) ;
511+ }
512+ }
0 commit comments