(
&self,
_database_id: nodedb_types::DatabaseId,
name: &str,
)
| 167 | |
| 168 | impl SqlCatalog for OriginCatalog { |
| 169 | fn get_collection( |
| 170 | &self, |
| 171 | _database_id: nodedb_types::DatabaseId, |
| 172 | name: &str, |
| 173 | ) -> std::result::Result<Option<CollectionInfo>, SqlCatalogError> { |
| 174 | // Read through the local `SystemCatalog` redb. On cluster |
| 175 | // followers, the `MetadataCommitApplier` has already |
| 176 | // written the replicated record here via |
| 177 | // `CatalogEntry::apply_to`, so a single read path works |
| 178 | // for both single-node and cluster modes. |
| 179 | // |
| 180 | // Use `self.database_id` (the session-bound database) rather than |
| 181 | // the `_database_id` parameter, which is always `DatabaseId::DEFAULT` |
| 182 | // from the nodedb-sql planner. This enforces per-database namespace |
| 183 | // isolation at plan time: a query in `db_alpha` cannot resolve a |
| 184 | // collection that lives in `db_beta`. |
| 185 | let catalog_ref = self.credentials.catalog(); |
| 186 | let Some(catalog) = catalog_ref.as_ref() else { |
| 187 | return Ok(None); |
| 188 | }; |
| 189 | let Some(stored) = catalog |
| 190 | .get_collection(self.database_id, self.tenant_id, name) |
| 191 | .ok() |
| 192 | .flatten() |
| 193 | else { |
| 194 | return Ok(None); |
| 195 | }; |
| 196 | if !stored.is_active { |
| 197 | // Soft-deleted: surface a distinct error so the pgwire |
| 198 | // handler renders the UNDROP hint instead of "unknown |
| 199 | // table". Retention window uses the default config — |
| 200 | // per-tenant override resolution is tracked as its own |
| 201 | // checklist item. |
| 202 | let retention = crate::config::server::RetentionSettings::default() |
| 203 | .retention_window() |
| 204 | .as_nanos() as u64; |
| 205 | let retention_expires_at_ns = stored.modification_hlc.wall_ns.saturating_add(retention); |
| 206 | return Err(SqlCatalogError::CollectionDeactivated { |
| 207 | name: name.to_string(), |
| 208 | retention_expires_at_ns, |
| 209 | }); |
| 210 | } |
| 211 | |
| 212 | // Record the observed descriptor version so the caller |
| 213 | // can use the resulting set as a per-descriptor plan |
| 214 | // cache key. The set is drained via |
| 215 | // `take_recorded_versions` once planning finishes. |
| 216 | // |
| 217 | // Version 0 is the pre-B.1 sentinel; we record it as 1 |
| 218 | // so the cache's freshness check uses the same floor |
| 219 | // that the drain gate uses. If the descriptor later |
| 220 | // stamps its first real version 1, the cache stays |
| 221 | // valid; if it bumps to 2+, the cache correctly |
| 222 | // invalidates. |
| 223 | let descriptor_id = DescriptorId::new( |
| 224 | self.tenant_id, |
| 225 | DescriptorKind::Collection, |
| 226 | stored.name.clone(), |
nothing calls this directly
no test coverage detected