(
&self,
task: PhysicalTask,
user_id: Option<Arc<str>>,
)
| 41 | } |
| 42 | |
| 43 | async fn dispatch_task_inner( |
| 44 | &self, |
| 45 | task: PhysicalTask, |
| 46 | user_id: Option<Arc<str>>, |
| 47 | ) -> crate::Result<Response> { |
| 48 | // Reject user writes against a source database that is currently |
| 49 | // frozen by a clone materializer sweep. Reads and DDL pass through |
| 50 | // unchanged. The materializer uses `dispatch_local` (a free function |
| 51 | // in `clone_materializer/dispatch.rs`) and is never routed through |
| 52 | // this method, so there is no risk of blocking the materializer itself. |
| 53 | use crate::control::security::identity::{Permission, required_permission}; |
| 54 | let perm = required_permission(&task.plan); |
| 55 | if matches!(perm, Permission::Write | Permission::Admin) |
| 56 | && self.state.materialize_freeze.is_frozen(task.database_id) |
| 57 | { |
| 58 | return Err(crate::Error::SourceFrozen { |
| 59 | database_id: task.database_id, |
| 60 | }); |
| 61 | } |
| 62 | |
| 63 | // Mirror enforcement: |
| 64 | // - Writes are rejected on non-promoted mirrors (MIRROR_READ_ONLY). |
| 65 | // - Reads are gated by the session's ReadConsistency level: |
| 66 | // Strong → STALE_READ_NOT_LEADER (mirrors are never the source leader) |
| 67 | // BoundedStaleness(d) → serve locally if lag ≤ d, else STALE_READ_NOT_LEADER |
| 68 | // Eventual → serve locally unconditionally |
| 69 | // The catalog lookup is skipped for the default database (id=0) to keep the |
| 70 | // hot path allocation-free in the single-database case. |
| 71 | if task.database_id.as_u64() != 0 |
| 72 | && let Some(catalog) = self.state.credentials.catalog() |
| 73 | && let Ok(Some(descriptor)) = catalog.get_database(task.database_id) |
| 74 | && let Some(origin) = descriptor.mirror_origin.as_ref() |
| 75 | && !matches!(origin.status, nodedb_types::MirrorStatus::Promoted) |
| 76 | { |
| 77 | if matches!(perm, Permission::Write | Permission::Admin) { |
| 78 | return Err(crate::Error::MirrorReadOnly { |
| 79 | database: descriptor.name.clone(), |
| 80 | }); |
| 81 | } |
| 82 | |
| 83 | use crate::control::server::pgwire::ddl::database::{ |
| 84 | MirrorReadOutcome, check_mirror_read_consistency, |
| 85 | }; |
| 86 | // Consistency defaults to Strong: mirrors are not the source leader, |
| 87 | // so reads are rejected unless the session has explicitly opted into |
| 88 | // BoundedStaleness or Eventual. |
| 89 | let outcome = check_mirror_read_consistency( |
| 90 | catalog, |
| 91 | task.database_id, |
| 92 | origin, |
| 93 | ReadConsistency::Strong, |
| 94 | ); |
| 95 | if let MirrorReadOutcome::Reject { message, .. } = outcome { |
| 96 | return Err(crate::Error::StaleReadNotLeader { |
| 97 | database: descriptor.name.clone(), |
| 98 | source_cluster: origin.source_cluster.clone(), |
| 99 | detail: message, |
| 100 | }); |
no test coverage detected