parallel-workload: prepare (disabled) extensions 2 and 4

Land the coverage for extension directions 2 and 4 but leave it commented out
of read_action_list, ready to flip on once the tracked bugs are fixed. Enabling
now would only re-detect known-unfixed coordinator/compute bugs, breaking
mergeability without finding anything new.

- DependencyConsistencyAction (ext 2, dependency oracle): flags any
  mz_object_dependencies edge whose endpoints are not both live objects, the
  SQL-521 MissingUses class. Enable once SQL-521 is fixed.

- SourceReadHoldSweepAction (ext 4, kill enrichment): reads source-backed
  relations to force read-hold acquisition on their remap shards, stressing
  reinstatement across restart (SS-346) and post-ALTER-TABLE-ADD-COLUMN as_of
  (PER-49). Enable once SS-346 and PER-49 are fixed.

Extension 3 (namespace CASCADE races) is already prepared this way:
DropDatabaseCascadeAction exists and is commented out with a SQL-518 TODO.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: def-

misc/python/materialize/parallel_workload/action.py

@@ -5309,6 +5309,85 @@ def run(self, exe: Executor) -> bool:
         return True
 
 
+class DependencyConsistencyAction(Action):
+    """Client-side catalog dependency oracle: flag any dependency edge whose
+    endpoints are not both live objects (a dangling uses/used_by edge).
+
+    PREPARED BUT DISABLED (commented out of read_action_list). This targets the
+    SQL-521 class: a sink left pointing at a materialized view that was dropped
+    under cancel, i.e. the same MissingUses inconsistency the coordinator's own
+    check_consistency asserts on. While SQL-521 is open the workload
+    legitimately produces such danglers, so enabling this now would re-detect
+    the known corruption rather than find new bugs.
+    TODO: enable in read_action_list once SQL-521 is fixed. Verify the
+    mz_internal.mz_object_dependencies column names (object_id,
+    referenced_object_id) still hold when enabling."""
+
+    def errors_to_ignore(self, exe: Executor) -> list[str]:
+        return [
+            # Reading a catalog relation inside a read txn that already touched
+            # user objects crosses timedomains.
+            "in the same timedomain",
+        ] + super().errors_to_ignore(exe)
+
+    def run(self, exe: Executor) -> bool:
+        exe.execute(
+            "SELECT d.object_id, d.referenced_object_id "
+            "FROM mz_internal.mz_object_dependencies d "
+            "LEFT JOIN mz_objects o1 ON d.object_id = o1.id "
+            "LEFT JOIN mz_objects o2 ON d.referenced_object_id = o2.id "
+            "WHERE o1.id IS NULL OR o2.id IS NULL",
+            http=Http.NO,
+        )
+        dangling = exe.cur.fetchall()
+        if dangling:
+            raise ValueError(
+                f"dangling catalog dependency edges (SQL-521 class): {dangling[:5]}"
+            )
+        return True
+
+
+class SourceReadHoldSweepAction(Action):
+    """Read a source-backed relation to force read-hold acquisition on the
+    source's remap shard.
+
+    PREPARED BUT DISABLED (commented out of read_action_list). Intended for the
+    kill scenario, where racing envd/clusterd restarts stresses read-hold
+    reinstatement: the SS-346 class (a dependent's read hold on a source's remap
+    shard not upheld across a restart, so its since advances past the
+    dependent's upper) and PER-49 (a compute import as_of behind the compacted
+    since after ALTER TABLE ADD COLUMN + kill). Enabling it now just re-triggers
+    those known coordinator/compute panics.
+    TODO: enable in read_action_list once SS-346 and PER-49 are fixed."""
+
+    def errors_to_ignore(self, exe: Executor) -> list[str]:
+        return [
+            "in the same timedomain",
+        ] + super().errors_to_ignore(exe)
+
+    def run(self, exe: Executor) -> bool:
+        with exe.db.lock:
+            sources = [
+                o
+                for o in exe.db.db_objects()
+                if isinstance(
+                    o,
+                    LoadGeneratorSource
+                    | KafkaSource
+                    | PostgresSource
+                    | MySqlSource
+                    | SqlServerSource
+                    | WebhookSource,
+                )
+            ]
+            if not sources:
+                return False
+            obj = self.rng.choice(sources)
+        exe.execute(f"SELECT count(*) FROM {obj}", http=Http.RANDOM)
+        exe.cur.fetchall()
+        return True
+
+
 class ActionList:
     action_classes: list[type[Action]]
     weights: list[float]
@@ -5342,6 +5421,10 @@ def __init__(
         # EXPLAIN ANALYZE handle the drop gracefully). See SQL-519 /
         # FINDINGS-BUGS.md.
         # (ExplainFilterPushdownAction, 5),
+        # PREPARED BUT DISABLED (see class docstrings): enabling these now just
+        # re-detects known-unfixed coordinator bugs rather than finding new ones.
+        # (DependencyConsistencyAction, 5),  # TODO: enable once SQL-521 fixed
+        # (SourceReadHoldSweepAction, 5),  # TODO: enable once SS-346 & PER-49 fixed
         (SetClusterAction, 1),
         (CommitRollbackAction, 30),
         (ReconnectAction, 1),