storage: write status updates in the controller, and add the "paused" status

misc/python/materialize/checks/all_checks/source_errors.py

@@ -75,6 +75,10 @@ def manipulate(self) -> list[Testdrive]:
                 $ postgres-execute connection=postgres://postgres:postgres@postgres
                 DROP PUBLICATION IF EXISTS source_errors_publicationA;
                 INSERT INTO source_errors_table VALUES (3);
+
+                # We sleep for a bit here to allow status updates to propagate to the storage controller
+                # in scenarios where environmentd is killed
+                $ sleep-is-probably-flaky-i-have-justified-my-need-with-a-comment duration=5s
                 """,
                 """
                 $ postgres-execute connection=postgres://postgres:postgres@postgres

src/storage-client/src/client.proto

@@ -12,6 +12,7 @@
 syntax = "proto3";
 
 import "proto/src/proto.proto";
+import "proto/src/chrono.proto";
 import "repr/src/antichain.proto";
 import "repr/src/global_id.proto";
 import "cluster-client/src/client.proto";
@@ -111,6 +112,19 @@ message ProtoStorageResponse {
         repeated ProtoSinkStatisticsUpdate sink_updates = 2;
     }
 
+    message ProtoStatusUpdate {
+        mz_repr.global_id.ProtoGlobalId id = 1;
+        string status = 2;
+        mz_proto.chrono.ProtoNaiveDateTime timestamp = 3;
+        optional string error = 4;
+        repeated string hints = 5;
+        map<string, string> namespaced_errors = 6;
+    }
+
+    message ProtoStatusUpdates {
+        repeated ProtoStatusUpdate updates = 1;
+    }
+
     message ProtoDroppedIds {
         repeated mz_repr.global_id.ProtoGlobalId ids = 1;
     }
@@ -119,5 +133,6 @@ message ProtoStorageResponse {
         ProtoFrontierUppersKind frontier_uppers = 1;
         ProtoDroppedIds dropped_ids = 2;
         ProtoStatisticsUpdates stats = 3;
+        ProtoStatusUpdates status_updates = 4;
     }
 }

src/storage-client/src/client.rs

@@ -354,6 +354,49 @@ impl PackableStats for SinkStatisticsUpdate {
     }
 }
 
+/// A source or sink status update.
+///
+/// Represents a status update for a given object type. The inner value for each
+/// variant should be able to be packed into a status row that conforms to the schema
+/// for the object's status history relation.
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+pub struct StatusUpdate {
+    pub id: GlobalId,
+    pub status: String,
+    pub timestamp: chrono::DateTime<chrono::Utc>,
+    pub error: Option<String>,
+    pub hints: BTreeSet<String>,
+    pub namespaced_errors: BTreeMap<String, String>,
+}
+
+impl RustType<proto_storage_response::ProtoStatusUpdate> for StatusUpdate {
+    fn into_proto(&self) -> proto_storage_response::ProtoStatusUpdate {
+        proto_storage_response::ProtoStatusUpdate {
+            id: Some(self.id.into_proto()),
+            status: self.status.clone(),
+            timestamp: Some(self.timestamp.into_proto()),
+            error: self.error.clone(),
+            hints: self.hints.iter().cloned().collect(),
+            namespaced_errors: self.namespaced_errors.clone(),
+        }
+    }
+
+    fn from_proto(
+        proto: proto_storage_response::ProtoStatusUpdate,
+    ) -> Result<Self, TryFromProtoError> {
+        Ok(StatusUpdate {
+            id: proto.id.into_rust_if_some("ProtoStatusUpdate::id")?,
+            timestamp: proto
+                .timestamp
+                .into_rust_if_some("ProtoStatusUpdate::timestamp")?,
+            status: proto.status,
+            error: proto.error,
+            hints: proto.hints.into_iter().collect(),
+            namespaced_errors: proto.namespaced_errors,
+        })
+    }
+}
+
 /// Responses that the storage nature of a worker/dataflow can provide back to the coordinator.
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
 pub enum StorageResponse<T = mz_repr::Timestamp> {
@@ -367,14 +410,17 @@ pub enum StorageResponse<T = mz_repr::Timestamp> {
 
     /// A list of statistics updates, currently only for sources.
     StatisticsUpdates(Vec<SourceStatisticsUpdate>, Vec<SinkStatisticsUpdate>),
+    /// A list of status updates for sources and sinks. Periodically sent from
+    /// storage workers to convey the latest status information about an object.
+    StatusUpdates(Vec<StatusUpdate>),
 }
 
 impl RustType<ProtoStorageResponse> for StorageResponse<mz_repr::Timestamp> {
     fn into_proto(&self) -> ProtoStorageResponse {
         use proto_storage_response::Kind::*;
         use proto_storage_response::{
             ProtoDroppedIds, ProtoSinkStatisticsUpdate, ProtoSourceStatisticsUpdate,
-            ProtoStatisticsUpdates,
+            ProtoStatisticsUpdates, ProtoStatusUpdates,
         };
         ProtoStorageResponse {
             kind: Some(match self {
@@ -412,13 +458,16 @@ impl RustType<ProtoStorageResponse> for StorageResponse<mz_repr::Timestamp> {
                             .collect(),
                     })
                 }
+                StorageResponse::StatusUpdates(updates) => StatusUpdates(ProtoStatusUpdates {
+                    updates: updates.into_proto(),
+                }),
             }),
         }
     }
 
     fn from_proto(proto: ProtoStorageResponse) -> Result<Self, TryFromProtoError> {
         use proto_storage_response::Kind::*;
-        use proto_storage_response::ProtoDroppedIds;
+        use proto_storage_response::{ProtoDroppedIds, ProtoStatusUpdates};
         match proto.kind {
             Some(DroppedIds(ProtoDroppedIds { ids })) => {
                 Ok(StorageResponse::DroppedIds(ids.into_rust()?))
@@ -464,6 +513,9 @@ impl RustType<ProtoStorageResponse> for StorageResponse<mz_repr::Timestamp> {
                     })
                     .collect::<Result<Vec<_>, TryFromProtoError>>()?,
             )),
+            Some(StatusUpdates(ProtoStatusUpdates { updates })) => {
+                Ok(StorageResponse::StatusUpdates(updates.into_rust()?))
+            }
             None => Err(TryFromProtoError::missing_field(
                 "ProtoStorageResponse::kind",
             )),
@@ -660,6 +712,9 @@ where
                     sink_stats,
                 )))
             }
+            StorageResponse::StatusUpdates(updates) => {
+                Some(Ok(StorageResponse::StatusUpdates(updates)))
+            }
         }
     }
 }