Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Temporal maintenance: the broker's own scheduled
3 : //! job replaces TimescaleDB background workers in plain mode, and drives the
4 : //! retention knob in both modes.
5 : //!
6 : //! Single-winner rule: the run is claimed via
7 : //! `SELECT … FOR UPDATE SKIP LOCKED` on the `maintenance_jobs` row — N
8 : //! instances race, one wins, the rest skip. No coordinator.
9 : //!
10 : //! Plain-mode partitioning: weekly partitions are pre-created for a window
11 : //! around now; everything else (historic backfill) lands in the DEFAULT
12 : //! partition. Creating a partition whose range already has rows sitting in
13 : //! the DEFAULT partition fails in PostgreSQL, so a single row written with an
14 : //! `observedAt` past the window permanently blocks that week's partition and
15 : //! sends all of its later traffic to DEFAULT as well. Such a range is
16 : //! recovered rather than skipped: the rows are moved out of DEFAULT into a
17 : //! standalone table, which is then ATTACHed as the partition.
18 : //!
19 : //! The recovery belongs here and NOT at ingest, for two reasons. The first is
20 : //! internal: clamping an `observedAt` outside a horizon would let the
21 : //! `observed_at` column disagree with the raw timestamp string in `data`, and
22 : //! `compile::temporal::column_range_bound` may prune on that column only
23 : //! because it is a superset of the byte-exact text window (4.11). The second
24 : //! is normative: 4.8 defines `observedAt` as "the temporal Property at which a
25 : //! certain Property or Relationship became valid or was observed" and requires
26 : //! only that it be a 4.6.3 DateTime — there is no horizon and no error type
27 : //! for one, and a forecast Property legitimately becomes valid in the future.
28 : //! A well-formed future `observedAt` is therefore valid input the broker
29 : //! stores, not input it may refuse.
30 : //!
31 : //! Residual, deliberate: rows whose `observed_at` lies beyond the pre-created
32 : //! window stay in DEFAULT until their week enters it, and rows dated far
33 : //! enough ahead stay there indefinitely — retention only purges DEFAULT rows
34 : //! that are already OLD. Auto-adopting arbitrary future weeks is not the fix
35 : //! either: one row per week over a century would trade an oversized DEFAULT
36 : //! for thousands of partitions. The condition is reported instead (see
37 : //! `default_partition_load`), so an operator sees it rather than discovering
38 : //! it as a query slowdown.
39 :
40 : use sqlx::postgres::PgPool;
41 : use sqlx::{Acquire, Row};
42 :
43 : /// True when the timescaledb extension is CREATED in this database
44 : /// (per-database `pg_extension`, not "installed on the server").
45 3 : pub async fn timescale_present(pool: &PgPool) -> Result<bool, sqlx::Error> {
46 3 : let row = sqlx::query("SELECT 1 FROM pg_extension WHERE extname = 'timescaledb'")
47 3 : .fetch_optional(pool)
48 3 : .await?;
49 3 : Ok(row.is_some())
50 3 : }
51 :
52 : /// What the migrations actually built `attr_instances` as. Detected ONCE at
53 : /// startup from the catalog and pinned — never re-probed per tick, so the
54 : /// maintenance branch can never disagree with the DDL on disk (the
55 : /// "extension installed after first boot" trap).
56 : #[derive(Clone, Copy, Debug, PartialEq, Eq)]
57 : pub enum TemporalBackend {
58 : /// timescale hypertable (relkind 'r' + timescaledb catalog entry)
59 : Hypertable,
60 : /// native PARTITION BY RANGE (relkind 'p')
61 : Partitioned,
62 : }
63 :
64 : /// Inspect the catalog. Errors when `attr_instances` is neither a hypertable
65 : /// nor a partitioned table — that means the database was migrated under one
66 : /// extension state and is now running under another; refusing beats running
67 : /// the wrong maintenance jobs against mismatched DDL.
68 24 : pub async fn detect_temporal_backend(pool: &PgPool) -> Result<TemporalBackend, String> {
69 24 : let relkind: Option<String> = sqlx::query_scalar(
70 24 : "SELECT c.relkind::text FROM pg_class c
71 24 : WHERE c.relname = 'attr_instances' AND c.relnamespace = 'public'::regnamespace",
72 24 : )
73 24 : .fetch_optional(pool)
74 24 : .await
75 24 : .map_err(|e| e.to_string())?;
76 24 : match relkind.as_deref() {
77 24 : Some("p") => Ok(TemporalBackend::Partitioned),
78 6 : Some("r") => {
79 6 : let hyper = sqlx::query(
80 6 : "SELECT 1 FROM timescaledb_information.hypertables
81 6 : WHERE hypertable_name = 'attr_instances'",
82 6 : )
83 6 : .fetch_optional(pool)
84 6 : .await
85 6 : .map_err(|e| e.to_string())?;
86 6 : if hyper.is_some() {
87 6 : Ok(TemporalBackend::Hypertable)
88 : } else {
89 0 : Err(
90 0 : "attr_instances is a plain table — the database was migrated as a \
91 0 : hypertable and the timescaledb extension has since been removed, or \
92 0 : the catalog is damaged; refusing to run temporal maintenance"
93 0 : .into(),
94 0 : )
95 : }
96 : }
97 0 : other => Err(format!(
98 0 : "attr_instances has unexpected relkind {other:?} — migrations did not run?"
99 0 : )),
100 : }
101 24 : }
102 :
103 : /// How long a published claim-check row outlives its message. The changes
104 : /// stream keeps a message until every durable has acked it, so no bound is
105 : /// derivable from the bus — this is the operator-visible ceiling on how long
106 : /// a matcher may lag before an oversized change stops resolving.
107 : // ponytail: fixed window, an env knob if a deployment's matcher lag exceeds it
108 : const CLAIM_CHECK_HOURS: i64 = 24;
109 :
110 : /// One maintenance pass. Returns a short human-readable summary ("skipped"
111 : /// when another instance holds the claim). `retention_days = None` keeps
112 : /// history forever — retention is a deliberate deployment knob, never a
113 : /// default.
114 1009 : pub async fn temporal_maintenance(
115 1009 : pool: &PgPool,
116 1009 : backend: TemporalBackend,
117 1009 : retention_days: Option<i64>,
118 1009 : ) -> Result<String, sqlx::Error> {
119 1009 : let mut tx = pool.begin().await?;
120 1009 : let claimed = sqlx::query(
121 1009 : "SELECT name FROM maintenance_jobs WHERE name = 'temporal_partitions'
122 1009 : FOR UPDATE SKIP LOCKED",
123 1009 : )
124 1009 : .fetch_optional(&mut *tx)
125 1009 : .await?;
126 1009 : if claimed.is_none() {
127 2 : return Ok("skipped: another instance holds the claim".into());
128 1007 : }
129 : // retention DML is cross-tenant service work (the `antares.service`
130 : // escape in 0001_init.sql)
131 1007 : crate::store::pg::set_service(&mut tx).await?;
132 1007 : let mut done: Vec<String> = Vec::new();
133 : // The 4.22 reaps run in their own transactions AFTER this one commits: both
134 : // DELETEs grow with stored volume and can exceed the connection's
135 : // statement_timeout, and a reap that times out must not abort the partition
136 : // pre-creation that keeps ingest writable.
137 1007 : if backend == TemporalBackend::Hypertable {
138 494 : if let Some(days) = retention_days {
139 0 : for sql in [
140 0 : "SELECT public.drop_chunks('attr_instances', older_than => make_interval(days => $1::int))",
141 0 : "SELECT public.drop_chunks('attr_rows', older_than => make_interval(days => $1::int))",
142 0 : ] {
143 0 : sqlx::query(sql).bind(days).execute(&mut *tx).await?;
144 : }
145 0 : done.push(format!("timescale drop_chunks older than {days}d"));
146 494 : }
147 : } else {
148 : // weekly partitions for [now-1w, now+4w); a week from the cut-over on
149 : // (migration 0020) is a week of attr_rows, every other one of
150 : // attr_instances
151 2565 : for off in -1i64..4 {
152 2565 : let row = sqlx::query(
153 2565 : "SELECT to_char(date_trunc('week', now()) + make_interval(weeks => $1::int), 'IYYY\"w\"IW') AS suffix,
154 2565 : (date_trunc('week', now()) + make_interval(weeks => $1::int))::text AS lo,
155 2565 : (date_trunc('week', now()) + make_interval(weeks => ($1::int) + 1))::text AS hi,
156 2565 : COALESCE(date_trunc('week', now()) + make_interval(weeks => $1::int)
157 2565 : >= (SELECT at FROM history_cutover), false) AS keyed",
158 2565 : )
159 2565 : .bind(off)
160 2565 : .fetch_one(&mut *tx)
161 2565 : .await?;
162 2565 : let (suffix, lo, hi): (String, String, String) =
163 2565 : (row.get("suffix"), row.get("lo"), row.get("hi"));
164 2565 : let keyed: bool = row.get("keyed");
165 : // The failure below is EXPECTED (see module docs), and in
166 : // PostgreSQL a failed statement aborts the whole transaction —
167 : // every later one then returns 25P02. Tolerating an error means
168 : // owning a savepoint to roll back to; without it the first
169 : // already-occupied range poisons the entire maintenance pass.
170 2565 : let mut sp = tx.begin().await?;
171 2565 : match sqlx::query(sqlx::AssertSqlSafe(create_partition_sql(
172 2565 : keyed, &suffix, &lo, &hi,
173 2565 : )))
174 2565 : .execute(&mut *sp)
175 2565 : .await
176 : {
177 : Ok(_) => {
178 2559 : sp.commit().await?;
179 2559 : done.push(format!("partition {suffix}: ok"));
180 2559 : police_partition(&mut tx, keyed, &suffix).await?;
181 2559 : continue;
182 : }
183 6 : Err(_) => sp.rollback().await?,
184 : }
185 : // Rows for this range already sit in DEFAULT. Adopt them: the move
186 : // empties the range, so the ATTACH's revalidation of DEFAULT
187 : // passes. Its ACCESS EXCLUSIVE lock is bounded by the connection's
188 : // lock_timeout, and a loser simply retries on the next tick.
189 6 : let mut sp = tx.begin().await?;
190 6 : let mut adopted = Ok(());
191 8 : for stmt in adopt_default_rows_sql(keyed, &suffix, &lo, &hi) {
192 8 : adopted = sqlx::query(sqlx::AssertSqlSafe(stmt))
193 8 : .execute(&mut *sp)
194 8 : .await
195 8 : .map(|_| ());
196 8 : if adopted.is_err() {
197 5 : break;
198 3 : }
199 : }
200 6 : match adopted {
201 : Ok(()) => {
202 1 : sp.commit().await?;
203 1 : done.push(format!("partition {suffix}: adopted from default"));
204 1 : police_partition(&mut tx, keyed, &suffix).await?;
205 : }
206 5 : Err(e) => {
207 : // warn, not debug: a PERMANENTLY failing create
208 : // (permissions) must be visible at default log level, or it
209 : // silently degrades to "everything lands in DEFAULT".
210 5 : sp.rollback().await?;
211 5 : tracing::warn!("weekly history partition {suffix} not created: {e}");
212 5 : done.push(format!("partition {suffix}: left in default"));
213 : }
214 : }
215 : }
216 : }
217 1007 : sqlx::query("UPDATE maintenance_jobs SET last_run = now() WHERE name = 'temporal_partitions'")
218 1007 : .execute(&mut *tx)
219 1007 : .await?;
220 1007 : tx.commit().await?;
221 : // Plain-mode retention runs AFTER the claim transaction commits, for the
222 : // same reason as the 4.22 reaps: DROP TABLE needs ACCESS EXCLUSIVE on the
223 : // parent and the DEFAULT purge grows with stored volume, so one lock
224 : // contention or one statement timeout would otherwise roll back the
225 : // partition pre-creation that keeps ingest writable.
226 : // The superseded versions live in one plain table beside the
227 : // partitions or chunks, and go with the platform retention all the same.
228 1007 : if let Some(days) = retention_days {
229 0 : match versions_retention(pool, days).await {
230 0 : Ok(0) => {}
231 0 : Ok(n) => done.push(format!(
232 : "deleted {n} superseded versions older than {days}d"
233 : )),
234 0 : Err(e) => done.push(format!("version retention skipped ({e})")),
235 : }
236 1007 : }
237 1007 : if backend == TemporalBackend::Partitioned {
238 513 : if let Some(days) = retention_days {
239 0 : match plain_retention(pool, days).await {
240 0 : Ok(lines) => done.extend(lines),
241 0 : Err(e) => done.push(format!("retention skipped ({e})")),
242 : }
243 513 : }
244 513 : match drop_emptied_partitions(pool).await {
245 513 : Ok(lines) => done.extend(lines),
246 0 : Err(e) => done.push(format!("emptied partitions kept ({e})")),
247 : }
248 513 : if let Ok(Some(line)) = default_partition_load(pool).await {
249 0 : done.push(line);
250 513 : }
251 513 : match history_one_index(pool).await {
252 1 : Ok(Some(line)) => done.push(line),
253 512 : Ok(None) => {}
254 0 : Err(e) => done.push(format!("history index build deferred ({e})")),
255 : }
256 494 : }
257 : // Before the reap that reads it: a week created ahead this pass is
258 : // indexed in this pass.
259 1007 : match history_expiry_index(pool).await {
260 15 : Ok(Some(line)) => done.push(line),
261 992 : Ok(None) => {}
262 0 : Err(e) => done.push(format!("history expiry index build deferred ({e})")),
263 : }
264 : // 4.22 garbage collection, on both backends: reads already refuse expired
265 : // entities and instances, so these reaps only bound storage — the clause
266 : // itself sanctions deletion lagging expiresAt.
267 1007 : match reap_expired_entities(pool).await {
268 991 : Ok(0) => {}
269 16 : Ok(n) => done.push(format!("reaped {n} expired transient entities (4.22)")),
270 0 : Err(e) => done.push(format!("entity reap skipped ({e})")),
271 : }
272 : // 4.22 also names Properties/Relationships: an attribute instance whose
273 : // expiresAt has passed "should be deleted from an NGSI-LD system". This
274 : // DELETE additionally contends with concurrent ingest on the same rows, and
275 : // a deadlock must cost only the reap (observed under ~1.2k msg/s ingest).
276 1007 : match reap_expired_instances(pool).await {
277 991 : Ok(0) => {}
278 15 : Ok(n) => done.push(format!("reaped {n} expired attribute instances (4.22)")),
279 1 : Err(e) => done.push(format!("instance reap skipped ({e})")),
280 : }
281 1007 : match expiry_scan_daily(pool).await {
282 1006 : Ok(None) | Ok(Some(0)) => {}
283 1 : Ok(Some(n)) => done.push(format!("daily scan reaped {n} expired history rows (4.22)")),
284 0 : Err(e) => done.push(format!("daily expiry scan failed ({e})")),
285 : }
286 : // Claim-check rows the drain kept because the bus could not carry their
287 : // bodies. The consumer resolves them off the message, so the window has to
288 : // outlive the message: reaping one early costs the notification it was
289 : // carrying, keeping one costs a row.
290 1007 : match crate::store::pg::outbox::reap_published(pool, CLAIM_CHECK_HOURS).await {
291 1007 : Ok(0) => {}
292 0 : Ok(n) => done.push(format!("reaped {n} published claim-check rows")),
293 0 : Err(e) => done.push(format!("claim-check reap skipped ({e})")),
294 : }
295 1007 : if done.is_empty() {
296 470 : done.push("nothing to do".into());
297 983 : }
298 1007 : Ok(done.join("; "))
299 1009 : }
300 :
301 : /// The superseded versions older than `days` days, on both backends.
302 0 : async fn versions_retention(pool: &PgPool, days: i64) -> Result<u64, sqlx::Error> {
303 0 : let mut tx = pool.begin().await?;
304 0 : crate::store::pg::set_service(&mut tx).await?;
305 0 : let n = sqlx::query(
306 0 : "DELETE FROM attr_versions WHERE observed_at < now() - make_interval(days => $1::int)",
307 0 : )
308 0 : .bind(days)
309 0 : .execute(&mut *tx)
310 0 : .await?
311 0 : .rows_affected();
312 0 : tx.commit().await?;
313 0 : Ok(n)
314 0 : }
315 :
316 : /// Plain-mode retention, in its own transaction: drop every weekly partition
317 : /// whose whole range is older than the horizon, then purge the DEFAULT
318 : /// partition (which has no upper bound and is therefore never dropped) of the
319 : /// historic-backfill rows that landed there before their week existed.
320 : ///
321 : /// Both statements are idempotent, so a run that loses a lock simply repeats
322 : /// on the next tick. Service role: retention is cross-tenant work.
323 0 : async fn plain_retention(pool: &PgPool, days: i64) -> Result<Vec<String>, sqlx::Error> {
324 0 : let mut done: Vec<String> = Vec::new();
325 0 : let mut tx = pool.begin().await?;
326 0 : crate::store::pg::set_service(&mut tx).await?;
327 0 : let parts = sqlx::query(
328 0 : // The parent is pinned to the schema the migrations built. Without
329 0 : // it a same-named table in another schema contributes its children,
330 0 : // and the DROP below would name them by bare relname.
331 0 : "SELECT c.relname,
332 0 : pg_get_expr(c.relpartbound, c.oid) AS bound
333 0 : FROM pg_inherits i
334 0 : JOIN pg_class c ON c.oid = i.inhrelid
335 0 : JOIN pg_class p ON p.oid = i.inhparent
336 0 : WHERE p.relname IN ('attr_instances', 'attr_rows_long')
337 0 : AND p.relnamespace = 'public'::regnamespace",
338 0 : )
339 0 : .fetch_all(&mut *tx)
340 0 : .await?;
341 0 : for r in parts {
342 0 : let name: String = r.get("relname");
343 0 : let bound: String = r.get::<Option<String>, _>("bound").unwrap_or_default();
344 : // bound looks like: FOR VALUES FROM ('<lo>') TO ('<hi>')
345 0 : let Some(hi) = bound
346 0 : .split("TO ('")
347 0 : .nth(1)
348 0 : .and_then(|s| s.split('\'').next())
349 : else {
350 0 : continue; // DEFAULT partition — never dropped
351 : };
352 0 : let expired: bool =
353 0 : sqlx::query_scalar("SELECT $1::timestamptz < now() - make_interval(days => $2::int)")
354 0 : .bind(hi)
355 0 : .bind(days)
356 0 : .fetch_one(&mut *tx)
357 0 : .await?;
358 0 : if expired {
359 0 : sqlx::query(sqlx::AssertSqlSafe(drop_partition_sql(&name)))
360 0 : .execute(&mut *tx)
361 0 : .await?;
362 0 : done.push(format!("dropped expired partition {name}"));
363 0 : }
364 : }
365 0 : let mut purged = 0u64;
366 0 : for sql in [
367 0 : "DELETE FROM attr_instances_default
368 0 : WHERE observed_at < now() - make_interval(days => $1::int)",
369 0 : "DELETE FROM attr_rows_default
370 0 : WHERE observed_at < now() - make_interval(days => $1::int)",
371 0 : ] {
372 0 : let n = sqlx::query(sql)
373 0 : .bind(days)
374 0 : .execute(&mut *tx)
375 0 : .await?
376 0 : .rows_affected();
377 0 : purged = purged.saturating_add(n);
378 : }
379 0 : tx.commit().await?;
380 0 : if purged > 0 {
381 0 : done.push(format!(
382 0 : "purged {purged} expired rows from DEFAULT partition"
383 0 : ));
384 0 : }
385 0 : Ok(done)
386 0 : }
387 :
388 : /// Rows the DEFAULT partition holds, and a warning once it stops being
389 : /// incidental. Every row there is one no weekly partition covers — historic
390 : /// backfill, or an `observedAt` dated beyond the pre-created window — so the
391 : /// count is what makes an unpartitioned pile visible before it shows up as a
392 : /// query that stopped pruning. `reltuples` is the planner's estimate, so this
393 : /// costs a catalog lookup rather than a scan of the pile it is measuring;
394 : /// `-1` means "never analysed", which is reported as nothing.
395 : const DEFAULT_PARTITION_WARN_ROWS: i64 = 100_000;
396 :
397 513 : async fn default_partition_load(pool: &PgPool) -> Result<Option<String>, sqlx::Error> {
398 513 : let rows: Option<f32> = sqlx::query_scalar(
399 513 : "SELECT sum(greatest(reltuples, 0))::real FROM pg_class
400 513 : WHERE relname IN ('attr_instances_default', 'attr_rows_default')
401 513 : AND relnamespace = 'public'::regnamespace",
402 513 : )
403 513 : .fetch_optional(pool)
404 513 : .await?;
405 513 : let est = rows.unwrap_or(-1.0) as i64;
406 513 : if est < DEFAULT_PARTITION_WARN_ROWS {
407 513 : return Ok(None);
408 0 : }
409 0 : tracing::warn!(
410 : "the DEFAULT history partitions hold ~{est} rows: instances are being written outside the \
411 : maintained partition window (historic backfill, or an observedAt far in the future); \
412 : queries over those ranges cannot prune"
413 : );
414 0 : Ok(Some(format!("default partition ~{est} rows")))
415 513 : }
416 :
417 : /// The 4.22 expired-entity DELETE, isolated so a reap that outruns
418 : /// `statement_timeout` costs only itself. Served by the partial index on
419 : /// `expires_at` (0001_init.sql) — without it this is a sequential scan of
420 : /// every entity in the deployment. Service role: the reap is cross-tenant work
421 : /// (RLS would hide other tenants' rows). An Entity its `expiresAt` ends takes
422 : /// its history with it, in the same transaction:
423 : /// its instances under the service role, its temporal row under its own
424 : /// Tenant, as `temporal_entities` keeps the plain tenant policy.
425 1007 : async fn reap_expired_entities(pool: &PgPool) -> Result<u64, sqlx::Error> {
426 1007 : let mut tx = pool.begin().await?;
427 1007 : crate::store::pg::set_service(&mut tx).await?;
428 1007 : let gone: Vec<(String, String)> = sqlx::query_as(
429 1007 : "DELETE FROM entities WHERE expires_at IS NOT NULL AND expires_at < now()
430 1007 : RETURNING tenant_id, id",
431 1007 : )
432 1007 : .fetch_all(&mut *tx)
433 1007 : .await?;
434 1007 : let mut by_tenant: std::collections::BTreeMap<&str, Vec<String>> =
435 1007 : std::collections::BTreeMap::new();
436 1007 : for (tenant, id) in &gone {
437 33 : by_tenant.entry(tenant).or_default().push(id.clone());
438 33 : }
439 1007 : for (tenant, ids) in by_tenant {
440 69 : for sql in [
441 23 : "DELETE FROM attr_instances WHERE tenant_id = $1 AND entity_id = ANY($2::text[])",
442 23 : "DELETE FROM attr_rows WHERE tenant_id = $1 AND entity_no IN \
443 23 : (SELECT i.id FROM iri_ids i WHERE i.tenant_id = $1 AND i.iri = ANY($2::text[]))",
444 23 : "DELETE FROM attr_versions WHERE tenant_id = $1 AND entity_id = ANY($2::text[])",
445 23 : ] {
446 69 : sqlx::query(sql)
447 69 : .bind(tenant)
448 69 : .bind(&ids)
449 69 : .execute(&mut *tx)
450 69 : .await?;
451 : }
452 23 : sqlx::query(crate::SET_TENANT_SQL)
453 23 : .bind(tenant)
454 23 : .execute(&mut *tx)
455 23 : .await?;
456 23 : sqlx::query("DELETE FROM temporal_entities WHERE tenant_id = $1 AND id = ANY($2::text[])")
457 23 : .bind(tenant)
458 23 : .bind(&ids)
459 23 : .execute(&mut *tx)
460 23 : .await?;
461 : }
462 1007 : tx.commit().await?;
463 1007 : Ok(u64::try_from(gone.len()).unwrap_or(u64::MAX))
464 1007 : }
465 :
466 : /// Retention of current state (`/q/retention`): at most `batch`
467 : /// Entities of one Tenant and type that carry no `expiresAt` (a writer's 4.22
468 : /// stamp wins) and were last modified before `before`. The subquery bounds
469 : /// the statement; `i_entities_types` serves the (tenant, type) match. Service
470 : /// role, like every reap here; the 4.22 reap's precedent holds: rows go, no
471 : /// notification is owed for a removal the client declared. `dry_run` counts
472 : /// the same rows and deletes none.
473 4 : pub async fn retain_current(
474 4 : pool: &PgPool,
475 4 : tenant: &str,
476 4 : type_iri: &str,
477 4 : before: &str,
478 4 : batch: usize,
479 4 : dry_run: bool,
480 4 : ) -> Result<u64, sqlx::Error> {
481 : const ROWS: &str = "SELECT tenant_id, id FROM entities
482 : WHERE tenant_id = $1 AND types @> $2::text[]
483 : AND expires_at IS NULL AND modified_at < $3::timestamptz
484 : LIMIT $4";
485 4 : retain(
486 4 : pool,
487 4 : ROWS,
488 4 : "DELETE FROM entities WHERE (tenant_id, id) IN",
489 4 : (tenant, &[type_iri.to_owned()], None, before, batch),
490 4 : dry_run,
491 4 : )
492 4 : .await
493 4 : }
494 :
495 : /// One bounded retention statement: `rows` selects at most `$4` keys of
496 : /// Tenant `$1` matching the names `$2` (and, given, leaving out the Entities
497 : /// `$5`) before `$3`; `delete` removes them, a dry run counts them.
498 75 : async fn retain(
499 75 : pool: &PgPool,
500 75 : rows: &str,
501 75 : delete: &str,
502 75 : (tenant, names, except_ids, before, batch): (&str, &[String], Option<&[String]>, &str, usize),
503 75 : dry_run: bool,
504 75 : ) -> Result<u64, sqlx::Error> {
505 75 : let sql = if dry_run {
506 10 : format!("SELECT count(*) FROM ({rows}) r")
507 : } else {
508 65 : format!("{delete} ({rows})")
509 : };
510 75 : let mut tx = pool.begin().await?;
511 75 : crate::store::pg::set_service(&mut tx).await?;
512 : // The rule's own Tenant as well: `temporal_entities` keeps the plain
513 : // tenant policy (no service escape), so without it the history JOIN sees
514 : // no row under the broker's role and deletes nothing.
515 75 : sqlx::query(crate::SET_TENANT_SQL)
516 75 : .bind(tenant)
517 75 : .execute(&mut *tx)
518 75 : .await?;
519 75 : if !dry_run {
520 : // One batch past a third of the sweep's budget (15 s, antares-ops
521 : // retention) is cancelled rather than holding a connection and the
522 : // database for minutes; the sweep logs it, goes on with the
523 : // other steps and retries it the next minute, so one slow (Tenant,
524 : // type) cannot spend the whole budget.
525 65 : sqlx::query("SET LOCAL statement_timeout = '5s'")
526 65 : .execute(&mut *tx)
527 65 : .await?;
528 10 : }
529 : // Both parts are constants of this module; every value is bound.
530 75 : let mut query = sqlx::query(sqlx::AssertSqlSafe(sql))
531 75 : .bind(tenant)
532 75 : .bind(names)
533 75 : .bind(antares_store::filter::canonical_datetime(before).into_owned())
534 75 : .bind(i64::try_from(batch).unwrap_or(i64::MAX));
535 75 : if let Some(ids) = except_ids {
536 9 : query = query.bind(ids);
537 66 : }
538 75 : let n = if dry_run {
539 10 : let count: i64 = sqlx::Row::get(&query.fetch_one(&mut *tx).await?, 0);
540 10 : u64::try_from(count).unwrap_or_default()
541 : } else {
542 65 : query.execute(&mut *tx).await?.rows_affected()
543 : };
544 75 : tx.commit().await?;
545 75 : Ok(n)
546 75 : }
547 :
548 : /// At most `$4` instances of the Entities `$2` of Tenant `$1` observed
549 : /// before `$3`, by the (tenant, entity) prefix of the history index.
550 : const ENTITY_ROWS: &str = "SELECT a.tenant_id, a.entity_id, a.attr_id, a.instance_id, a.observed_at
551 : FROM attr_instances a
552 : WHERE a.tenant_id = $1 AND a.entity_id = ANY($2::text[])
553 : AND a.observed_at < $3::timestamptz
554 : LIMIT $4";
555 : /// The same over `attr_rows`, its keys mapped through the Tenant's
556 : /// `iri_ids`; both of its children, the short one included.
557 : const KEYED_ROWS: &str = "SELECT a.tenant_id, a.entity_no, a.attr_no, a.instance_no, a.observed_at
558 : FROM attr_rows a
559 : WHERE a.tenant_id = $1
560 : AND a.entity_no IN (SELECT i.id FROM iri_ids i WHERE i.tenant_id = $1 AND i.iri = ANY($2::text[]))
561 : AND a.observed_at < $3::timestamptz
562 : LIMIT $4";
563 : /// The same over the superseded versions.
564 : const VERSION_ROWS: &str =
565 : "SELECT a.tenant_id, a.entity_id, a.attr_id, a.instance_id, a.observed_at
566 : FROM attr_versions a
567 : WHERE a.tenant_id = $1 AND a.entity_id = ANY($2::text[])
568 : AND a.observed_at < $3::timestamptz
569 : LIMIT $4";
570 :
571 : /// Retention of history (`/q/retention`): at most `batch`
572 : /// attribute instances of the Entities `scope` names in one Tenant, observed
573 : /// before `before`. Types live on `temporal_entities`; the instance key is
574 : /// the unique index's, so the IN matches exactly the rows selected in either
575 : /// backend. `observed_at` prunes the partitions to the old weeks.
576 18 : pub async fn retain_history(
577 18 : pool: &PgPool,
578 18 : tenant: &str,
579 18 : scope: antares_store::HistoryScope<'_>,
580 18 : before: &str,
581 18 : batch: usize,
582 18 : dry_run: bool,
583 18 : ) -> Result<u64, sqlx::Error> {
584 : use antares_store::HistoryScope;
585 : const ENTITY: &str = ENTITY_ROWS;
586 : const TYPE: &str = "SELECT a.tenant_id, a.entity_id, a.attr_id, a.instance_id, a.observed_at
587 : FROM attr_instances a
588 : JOIN temporal_entities t ON t.tenant_id = a.tenant_id AND t.id = a.entity_id
589 : WHERE a.tenant_id = $1 AND t.types @> $2::text[]
590 : AND a.entity_id <> ALL($5::text[])
591 : AND a.observed_at < $3::timestamptz
592 : LIMIT $4";
593 : const REST: &str = "SELECT a.tenant_id, a.entity_id, a.attr_id, a.instance_id, a.observed_at
594 : FROM attr_instances a
595 : JOIN temporal_entities t ON t.tenant_id = a.tenant_id AND t.id = a.entity_id
596 : WHERE a.tenant_id = $1 AND NOT (t.types && $2::text[])
597 : AND a.entity_id <> ALL($5::text[])
598 : AND a.observed_at < $3::timestamptz
599 : LIMIT $4";
600 : // The type and rest selections over `attr_rows`, its keys mapped
601 : // through the Tenant's `iri_ids`; KEYED_ROWS is the Entity one.
602 : const R_TYPE: &str = "SELECT a.tenant_id, a.entity_no, a.attr_no, a.instance_no, a.observed_at
603 : FROM attr_rows a
604 : JOIN iri_ids i ON i.tenant_id = a.tenant_id AND i.id = a.entity_no
605 : JOIN temporal_entities t ON t.tenant_id = i.tenant_id AND t.id = i.iri
606 : WHERE a.tenant_id = $1 AND t.types @> $2::text[]
607 : AND i.iri <> ALL($5::text[])
608 : AND a.observed_at < $3::timestamptz
609 : LIMIT $4";
610 : const R_REST: &str = "SELECT a.tenant_id, a.entity_no, a.attr_no, a.instance_no, a.observed_at
611 : FROM attr_rows a
612 : JOIN iri_ids i ON i.tenant_id = a.tenant_id AND i.id = a.entity_no
613 : JOIN temporal_entities t ON t.tenant_id = i.tenant_id AND t.id = i.iri
614 : WHERE a.tenant_id = $1 AND NOT (t.types && $2::text[])
615 : AND i.iri <> ALL($5::text[])
616 : AND a.observed_at < $3::timestamptz
617 : LIMIT $4";
618 : // A measurement past retention takes its superseded versions along:
619 : // the same three selections over `attr_versions`, written out so
620 : // every statement on `attr_instances` stays a literal the tenant guard
621 : // reads (tests/attr_instances_tenant_predicate.rs).
622 : const V_ENTITY: &str = VERSION_ROWS;
623 : const V_TYPE: &str = "SELECT a.tenant_id, a.entity_id, a.attr_id, a.instance_id, a.observed_at
624 : FROM attr_versions a
625 : JOIN temporal_entities t ON t.tenant_id = a.tenant_id AND t.id = a.entity_id
626 : WHERE a.tenant_id = $1 AND t.types @> $2::text[]
627 : AND a.entity_id <> ALL($5::text[])
628 : AND a.observed_at < $3::timestamptz
629 : LIMIT $4";
630 : const V_REST: &str = "SELECT a.tenant_id, a.entity_id, a.attr_id, a.instance_id, a.observed_at
631 : FROM attr_versions a
632 : JOIN temporal_entities t ON t.tenant_id = a.tenant_id AND t.id = a.entity_id
633 : WHERE a.tenant_id = $1 AND NOT (t.types && $2::text[])
634 : AND a.entity_id <> ALL($5::text[])
635 : AND a.observed_at < $3::timestamptz
636 : LIMIT $4";
637 : // A sweep's delete of a type walks its Entities: see
638 : // [`walk_history`]. A dry run counts the whole scope in one statement.
639 18 : if !dry_run {
640 : let one;
641 15 : let (ids, key, names, except) = match scope {
642 2 : HistoryScope::Entity(_) => (None, String::new(), &[][..], &[][..]),
643 : HistoryScope::Type {
644 12 : type_iri,
645 12 : except_ids,
646 : } => {
647 12 : one = [type_iri.to_owned()];
648 12 : (
649 12 : Some(TYPE_IDS),
650 12 : format!("{tenant}\n{type_iri}"),
651 12 : &one[..],
652 12 : except_ids,
653 12 : )
654 : }
655 : HistoryScope::Rest {
656 1 : except_types,
657 1 : except_ids,
658 1 : } => (
659 1 : Some(REST_IDS),
660 1 : format!("{tenant}\n"),
661 1 : except_types,
662 1 : except_ids,
663 1 : ),
664 : };
665 15 : if let Some(ids) = ids {
666 13 : let walk = Walk {
667 13 : ids,
668 13 : key,
669 13 : names,
670 13 : except,
671 13 : };
672 13 : return walk_history(pool, tenant, &walk, before, batch).await;
673 2 : }
674 3 : }
675 : let one;
676 5 : let (rows, keyed, versions, names, except) = match scope {
677 2 : HistoryScope::Entity(id) => {
678 2 : one = [id.to_owned()];
679 2 : (ENTITY, KEYED_ROWS, V_ENTITY, &one[..], None)
680 : }
681 : HistoryScope::Type {
682 2 : type_iri,
683 2 : except_ids,
684 : } => {
685 2 : one = [type_iri.to_owned()];
686 2 : (TYPE, R_TYPE, V_TYPE, &one[..], Some(except_ids))
687 : }
688 : HistoryScope::Rest {
689 1 : except_types,
690 1 : except_ids,
691 1 : } => (REST, R_REST, V_REST, except_types, Some(except_ids)),
692 : };
693 5 : let current = retain(
694 5 : pool,
695 5 : rows,
696 5 : "DELETE FROM attr_instances
697 5 : WHERE (tenant_id, entity_id, attr_id, instance_id, observed_at) IN",
698 5 : (tenant, names, except, before, batch),
699 5 : dry_run,
700 5 : )
701 5 : .await?;
702 5 : let current_keyed = retain(
703 5 : pool,
704 5 : keyed,
705 5 : "DELETE FROM attr_rows
706 5 : WHERE (tenant_id, entity_no, attr_no, instance_no, observed_at) IN",
707 5 : (tenant, names, except, before, batch),
708 5 : dry_run,
709 5 : )
710 5 : .await?;
711 5 : let superseded = retain(
712 5 : pool,
713 5 : versions,
714 5 : "DELETE FROM attr_versions
715 5 : WHERE (tenant_id, entity_id, attr_id, instance_id, observed_at) IN",
716 5 : (tenant, names, except, before, batch),
717 5 : dry_run,
718 5 : )
719 5 : .await?;
720 5 : Ok(current
721 5 : .saturating_add(current_keyed)
722 5 : .saturating_add(superseded))
723 18 : }
724 :
725 : /// The Entities of a type (`$2`), or of no type of `$2` for the rest, in
726 : /// Tenant `$1` leaving out `$3`, after `$4` in id order, at most `$5`. The
727 : /// primary key serves the order; `types` is only a filter, as no array
728 : /// operator reaches an index under RLS (migration 0015).
729 : const TYPE_IDS: &str = "SELECT t.id FROM temporal_entities t
730 : WHERE t.tenant_id = $1 AND t.types @> $2::text[]
731 : AND t.id <> ALL($3::text[]) AND t.id > $4
732 : ORDER BY t.id LIMIT $5";
733 : const REST_IDS: &str = "SELECT t.id FROM temporal_entities t
734 : WHERE t.tenant_id = $1 AND NOT (t.types && $2::text[])
735 : AND t.id <> ALL($3::text[]) AND t.id > $4
736 : ORDER BY t.id LIMIT $5";
737 :
738 : /// How long one [`walk_history`] call goes on to the next chunk of Entities
739 : /// when the ones before held nothing to delete.
740 : const WALK_TIME: std::time::Duration = std::time::Duration::from_secs(1);
741 :
742 : /// Where the walk of each (Tenant, scope) stopped, by [`Walk::key`]: the
743 : /// next call, the next sweep included, starts after it. Per process; a
744 : /// restart walks from the start again.
745 : static WALKS: std::sync::LazyLock<std::sync::Mutex<std::collections::HashMap<String, String>>> =
746 : std::sync::LazyLock::new(Default::default);
747 :
748 : /// One history retention scope as a walk over its Entities.
749 : struct Walk<'a> {
750 : ids: &'static str,
751 : key: String,
752 : names: &'a [String],
753 : except: &'a [String],
754 : }
755 :
756 : impl Walk<'_> {
757 13 : fn at(&self) -> String {
758 13 : let walks = WALKS
759 13 : .lock()
760 13 : .unwrap_or_else(std::sync::PoisonError::into_inner);
761 13 : walks.get(&self.key).cloned().unwrap_or_default()
762 13 : }
763 :
764 13 : fn stop(&self, after: String) {
765 13 : let mut walks = WALKS
766 13 : .lock()
767 13 : .unwrap_or_else(std::sync::PoisonError::into_inner);
768 13 : walks.insert(self.key.clone(), after);
769 13 : }
770 : }
771 :
772 : /// History retention of a type (or the rest) as a walk over its Entities in
773 : /// id order: each statement deletes at most `batch` instances of a
774 : /// chunk of `batch / 10` (10 to 100) Entities by the per-Entity index, so its
775 : /// cost is bounded by the chunk and never by the plan a JOIN of the type to
776 : /// the whole history gets. On a deployment that JOIN ran 4-10 s for a step with
777 : /// nothing to delete and timed out every few minutes. A chunk is left when
778 : /// it holds less than what is left of the batch; the walk goes on for up to
779 : /// [`WALK_TIME`] and remembers where it stopped, so a type of 50 000
780 : /// Entities is covered over a few sweeps instead of timing out in one. At
781 : /// the end it starts over. Returns the instances of both history parents
782 : /// and the versions deleted, at most `batch`.
783 13 : async fn walk_history(
784 13 : pool: &PgPool,
785 13 : tenant: &str,
786 13 : walk: &Walk<'_>,
787 13 : before: &str,
788 13 : batch: usize,
789 13 : ) -> Result<u64, sqlx::Error> {
790 13 : let started = std::time::Instant::now();
791 13 : let per = (batch / 10).clamp(10, 100);
792 13 : let mut after = walk.at();
793 13 : let mut total = 0usize;
794 : loop {
795 22 : let ids = entity_chunk(pool, tenant, walk, &after, per).await?;
796 22 : let Some(last) = ids.last().cloned() else {
797 0 : walk.stop(String::new());
798 0 : return Ok(total as u64);
799 : };
800 61 : for (rows, delete) in [
801 22 : (
802 22 : ENTITY_ROWS,
803 22 : "DELETE FROM attr_instances
804 22 : WHERE (tenant_id, entity_id, attr_id, instance_id, observed_at) IN",
805 22 : ),
806 22 : (
807 22 : KEYED_ROWS,
808 22 : "DELETE FROM attr_rows
809 22 : WHERE (tenant_id, entity_no, attr_no, instance_no, observed_at) IN",
810 22 : ),
811 22 : (
812 22 : VERSION_ROWS,
813 22 : "DELETE FROM attr_versions
814 22 : WHERE (tenant_id, entity_id, attr_id, instance_id, observed_at) IN",
815 22 : ),
816 22 : ] {
817 61 : let left = batch.saturating_sub(total);
818 61 : if left == 0 {
819 5 : break;
820 56 : }
821 56 : let n = retain(
822 56 : pool,
823 56 : rows,
824 56 : delete,
825 56 : (tenant, &ids, None, before, left),
826 56 : false,
827 56 : )
828 56 : .await?;
829 56 : total = total.saturating_add(usize::try_from(n).unwrap_or(usize::MAX));
830 : }
831 22 : if total >= batch {
832 : // The chunk may hold more: the next call starts with it again.
833 5 : walk.stop(after);
834 5 : return Ok(total as u64);
835 17 : }
836 17 : if ids.len() < per {
837 8 : walk.stop(String::new());
838 8 : return Ok(total as u64);
839 9 : }
840 9 : after = last;
841 9 : if started.elapsed() >= WALK_TIME {
842 0 : walk.stop(after);
843 0 : return Ok(total as u64);
844 9 : }
845 : }
846 13 : }
847 :
848 : /// The next chunk of a [`Walk`]'s Entities after `after`, under the rule's
849 : /// Tenant (`temporal_entities` has no service escape) and the same 5 s
850 : /// statement timeout as each delete.
851 22 : async fn entity_chunk(
852 22 : pool: &PgPool,
853 22 : tenant: &str,
854 22 : walk: &Walk<'_>,
855 22 : after: &str,
856 22 : per: usize,
857 22 : ) -> Result<Vec<String>, sqlx::Error> {
858 22 : let mut tx = pool.begin().await?;
859 22 : crate::store::pg::set_service(&mut tx).await?;
860 22 : sqlx::query(crate::SET_TENANT_SQL)
861 22 : .bind(tenant)
862 22 : .execute(&mut *tx)
863 22 : .await?;
864 22 : sqlx::query("SET LOCAL statement_timeout = '5s'")
865 22 : .execute(&mut *tx)
866 22 : .await?;
867 22 : let ids = sqlx::query_scalar(walk.ids)
868 22 : .bind(tenant)
869 22 : .bind(walk.names)
870 22 : .bind(walk.except)
871 22 : .bind(after)
872 22 : .bind(i64::try_from(per).unwrap_or(i64::MAX))
873 22 : .fetch_all(&mut *tx)
874 22 : .await?;
875 22 : tx.commit().await?;
876 22 : Ok(ids)
877 22 : }
878 :
879 : /// The Entities of every Tenant that carry their own history retention,
880 : /// served by the partial index of migration 0010 on the literal
881 : /// key, so the sweep reads them without scanning `entities`.
882 1 : pub async fn history_overrides(
883 1 : pool: &PgPool,
884 1 : limit: usize,
885 1 : ) -> Result<Vec<(String, String, serde_json::Value)>, sqlx::Error> {
886 1 : let mut tx = pool.begin().await?;
887 1 : crate::store::pg::set_service(&mut tx).await?;
888 1 : let rows = sqlx::query(
889 1 : "SELECT tenant_id, id, entity -> 'https://uri.etsi.org/ngsi-ld/default-context/historyRetention' AS own
890 1 : FROM entities
891 1 : WHERE entity ? 'https://uri.etsi.org/ngsi-ld/default-context/historyRetention'
892 1 : LIMIT $1",
893 1 : )
894 1 : .bind(i64::try_from(limit).unwrap_or(i64::MAX))
895 1 : .fetch_all(&mut *tx)
896 1 : .await?;
897 1 : tx.commit().await?;
898 1 : Ok(rows
899 1 : .into_iter()
900 2 : .map(|r| (r.get("tenant_id"), r.get("id"), r.get("own")))
901 1 : .collect())
902 1 : }
903 :
904 : /// Drop every weekly partition below the pre-created window that holds no
905 : /// row, so the disk the retention sweeps freed goes back to the node:
906 : /// rows deleted from a partition that keeps one free nothing, as new
907 : /// writes land in the current week. See [`drop_partitions`].
908 513 : async fn drop_emptied_partitions(pool: &PgPool) -> Result<Vec<String>, sqlx::Error> {
909 513 : drop_partitions(pool, &[], &[]).await
910 513 : }
911 :
912 : /// Retention by partition: drop every weekly partition whose rows
913 : /// all belong to Tenants of `floors` (Tenant, UTC DateTime) and whose upper
914 : /// bound is at or below that Tenant's floor, the oldest cut any of its rules
915 : /// keeps. The rows such a partition holds are all ones the row sweep would
916 : /// delete, so dropping the table replaces millions of row DELETEs, their WAL
917 : /// and the vacuum after them (one deployment: P28D deleted ~2M rows a night
918 : /// and starved the broker). A partition holding one row of a Tenant without a
919 : /// floor, or with a floor below its bound, stays for the row sweep.
920 3 : pub async fn drop_retained_partitions(
921 3 : pool: &PgPool,
922 3 : floors: &[(String, String)],
923 3 : ) -> Result<Vec<String>, sqlx::Error> {
924 3 : if floors.is_empty() {
925 1 : return Ok(Vec::new());
926 2 : }
927 2 : let (tenants, cuts): (Vec<String>, Vec<String>) = floors
928 2 : .iter()
929 4 : .map(|(t, c)| {
930 4 : (
931 4 : t.clone(),
932 4 : antares_store::filter::canonical_datetime(c).into_owned(),
933 4 : )
934 4 : })
935 2 : .unzip();
936 2 : drop_partitions(pool, &tenants, &cuts).await
937 3 : }
938 :
939 : /// Whether partition `{p}` may go: every Tenant with a row in it has a floor
940 : /// in (`$1` Tenants, `$2` cuts) at or above its upper bound `$3`. The
941 : /// Tenants present are read by a loose scan of the partition's
942 : /// `(tenant_id, ...)` index, one descent per Tenant, never the rows.
943 : const DROPPABLE: &str = "WITH RECURSIVE t(id) AS (
944 : (SELECT tenant_id FROM {p} ORDER BY tenant_id LIMIT 1)
945 : UNION ALL
946 : SELECT (SELECT tenant_id FROM {p} WHERE tenant_id > t.id ORDER BY tenant_id LIMIT 1)
947 : FROM t WHERE t.id IS NOT NULL)
948 : SELECT NOT EXISTS (
949 : SELECT 1 FROM t
950 : LEFT JOIN unnest($1::text[], $2::timestamptz[]) AS f(tenant, cut) ON f.tenant = t.id
951 : WHERE t.id IS NOT NULL AND (f.cut IS NULL OR f.cut < $3::timestamptz))";
952 :
953 : /// Drop the past weekly partitions [`DROPPABLE`] admits under (`tenants`,
954 : /// `cuts`); with none, the empty ones below the pre-created window. One
955 : /// transaction locks the parent, and with it every partition, `ACCESS
956 : /// EXCLUSIVE` within a 2 s `lock_timeout` and checks each candidate again
957 : /// under that lock, each check held to a 2 s `statement_timeout`, so a late
958 : /// write into the week either committed before (and is counted) or waits and
959 : /// lands in DEFAULT once the week is gone, and no history read or write
960 : /// waits longer than both bounds; a lock not had in time leaves it all for
961 : /// the next tick. Nothing to drop takes no lock.
962 515 : async fn drop_partitions(
963 515 : pool: &PgPool,
964 515 : tenants: &[String],
965 515 : cuts: &[String],
966 515 : ) -> Result<Vec<String>, sqlx::Error> {
967 : // A candidate ends at or below the latest floor, or below the window.
968 : const OLD_PARTITIONS: &str = r"SELECT c.relname, b.hi::text AS hi
969 : FROM pg_inherits i
970 : JOIN pg_class c ON c.oid = i.inhrelid
971 : JOIN pg_class p ON p.oid = i.inhparent
972 : CROSS JOIN LATERAL (SELECT substring(pg_get_expr(c.relpartbound, c.oid)
973 : FROM 'TO \(''([^'']+)''\)')::timestamptz AS hi) b
974 : WHERE p.relname IN ('attr_instances', 'attr_rows_long')
975 : AND p.relnamespace = 'public'::regnamespace
976 : AND pg_get_expr(c.relpartbound, c.oid) LIKE 'FOR VALUES FROM%'
977 : AND b.hi <= GREATEST((SELECT max(x) FROM unnest($1::timestamptz[]) x),
978 : date_trunc('week', now()) - interval '1 week')
979 : ORDER BY b.hi";
980 515 : let mut done = Vec::new();
981 : // Unlocked first, each check in a transaction of its own: one that held
982 : // a partition's lock while waiting on the next deadlocks with another
983 : // pod's LOCK below, which takes them all in turn.
984 515 : let mut tx = pool.begin().await?;
985 515 : let found: Vec<(String, String)> = sqlx::query_as(OLD_PARTITIONS)
986 515 : .bind(cuts)
987 515 : .fetch_all(&mut *tx)
988 515 : .await?;
989 515 : tx.commit().await?;
990 515 : let mut any = false;
991 2555 : for (name, hi) in &found {
992 2555 : let mut tx = pool.begin().await?;
993 2555 : crate::store::pg::set_service(&mut tx).await?;
994 2555 : let ok = droppable(&mut tx, name, hi, tenants, cuts).await?;
995 2555 : tx.commit().await?;
996 2555 : if ok {
997 4 : any = true;
998 4 : break;
999 2551 : }
1000 : }
1001 515 : if !any {
1002 511 : return Ok(done);
1003 4 : }
1004 4 : let mut tx = pool.begin().await?;
1005 : // The maintenance claim: one pod drops at a time, and a writer that holds
1006 : // it (a pass creating a week) never sees its new week dropped while empty.
1007 4 : let claimed = sqlx::query(
1008 4 : "SELECT name FROM maintenance_jobs WHERE name = 'temporal_partitions'
1009 4 : FOR UPDATE SKIP LOCKED",
1010 4 : )
1011 4 : .fetch_optional(&mut *tx)
1012 4 : .await?;
1013 4 : if claimed.is_none() {
1014 0 : return Ok(done);
1015 4 : }
1016 4 : crate::store::pg::set_service(&mut tx).await?;
1017 4 : sqlx::query("SET LOCAL lock_timeout = '2s'")
1018 4 : .execute(&mut *tx)
1019 4 : .await?;
1020 4 : sqlx::query("SET LOCAL statement_timeout = '2s'")
1021 4 : .execute(&mut *tx)
1022 4 : .await?;
1023 4 : sqlx::query("LOCK TABLE public.attr_instances, public.attr_rows IN ACCESS EXCLUSIVE MODE")
1024 4 : .execute(&mut *tx)
1025 4 : .await?;
1026 : // Read again under the lock: another pod may have dropped one meanwhile.
1027 4 : let found: Vec<(String, String)> = sqlx::query_as(OLD_PARTITIONS)
1028 4 : .bind(cuts)
1029 4 : .fetch_all(&mut *tx)
1030 4 : .await?;
1031 28 : for (name, hi) in found {
1032 28 : if droppable(&mut tx, &name, &hi, tenants, cuts).await? {
1033 4 : sqlx::query(sqlx::AssertSqlSafe(drop_partition_sql(&name)))
1034 4 : .execute(&mut *tx)
1035 4 : .await?;
1036 4 : done.push(format!("dropped partition {name}"));
1037 24 : }
1038 : }
1039 4 : tx.commit().await?;
1040 4 : Ok(done)
1041 515 : }
1042 :
1043 : /// Whether [`DROPPABLE`] admits the partition `name` with upper bound `hi`.
1044 : /// The partition is read directly, under its own policies (migration
1045 : /// 0029): the caller's transaction holds the service escape, so the read
1046 : /// sees every Tenant's rows.
1047 2583 : async fn droppable(
1048 2583 : tx: &mut sqlx::PgConnection,
1049 2583 : name: &str,
1050 2583 : hi: &str,
1051 2583 : tenants: &[String],
1052 2583 : cuts: &[String],
1053 2583 : ) -> Result<bool, sqlx::Error> {
1054 2583 : let quoted = format!("public.\"{}\"", name.replace('"', "\"\""));
1055 2583 : sqlx::query_scalar(sqlx::AssertSqlSafe(DROPPABLE.replace("{p}", "ed)))
1056 2583 : .bind(tenants)
1057 2583 : .bind(cuts)
1058 2583 : .bind(hi)
1059 2583 : .fetch_one(tx)
1060 2583 : .await
1061 2583 : }
1062 :
1063 : /// Drop one partition, named the way the catalog row spells it. The name
1064 : /// comes back from `pg_class.relname` as the identifier itself, not as SQL:
1065 : /// unquoted it would be folded to lower case and would need to be a bare
1066 : /// identifier to parse at all, so a partition whose name is neither — one
1067 : /// created by hand, or by a migration that quoted it — is a partition
1068 : /// retention fails on, every tick, for as long as it exists. Schema-
1069 : /// qualified for the same reason the catalog query is: the row names a table
1070 : /// in `public`, and `search_path` is not what should decide which one.
1071 10 : fn drop_partition_sql(name: &str) -> String {
1072 10 : format!("DROP TABLE public.\"{}\"", name.replace('"', "\"\""))
1073 10 : }
1074 :
1075 : /// One weekly partition, created directly under the parent: `attr_rows_long`,
1076 : /// the logged child of `attr_rows` (0023), when `keyed`, else
1077 : /// `attr_instances`. Fails while the parent's DEFAULT
1078 : /// partition still holds a row in `[lo, hi)`.
1079 2569 : fn create_partition_sql(keyed: bool, suffix: &str, lo: &str, hi: &str) -> String {
1080 2569 : if keyed {
1081 2 : format!(
1082 : "CREATE TABLE IF NOT EXISTS attr_rows_{suffix} PARTITION OF attr_rows_long \
1083 : FOR VALUES FROM ('{lo}') TO ('{hi}')"
1084 : )
1085 : } else {
1086 2567 : format!(
1087 : "CREATE TABLE IF NOT EXISTS attr_instances_{suffix} PARTITION OF attr_instances \
1088 : FOR VALUES FROM ('{lo}') TO ('{hi}')"
1089 : )
1090 : }
1091 2569 : }
1092 :
1093 : /// The week `suffix` given its parent's row security
1094 : /// (`history_partition_rls`, migration 0029): a statement naming the
1095 : /// partition is judged by the partition's own policies, never the parent's.
1096 : /// After the week is attached, so the adoption's move into the standalone
1097 : /// table is not refused by its WITH CHECK. In a savepoint of its own: a
1098 : /// week that cannot be policed (a role that does not own it) stays as it
1099 : /// was and is reported, it does not undo the week.
1100 2560 : async fn police_partition(
1101 2560 : tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
1102 2560 : keyed: bool,
1103 2560 : suffix: &str,
1104 2560 : ) -> Result<(), sqlx::Error> {
1105 2560 : let mut sp = tx.begin().await?;
1106 2560 : match sqlx::query(
1107 : "SELECT history_partition_rls(format('public.%I',
1108 : CASE WHEN $1 THEN 'attr_rows_' ELSE 'attr_instances_' END || $2)::regclass)",
1109 : )
1110 2560 : .bind(keyed)
1111 2560 : .bind(suffix)
1112 2560 : .execute(&mut *sp)
1113 2560 : .await
1114 : {
1115 2560 : Ok(_) => sp.commit().await,
1116 0 : Err(e) => {
1117 0 : sp.rollback().await?;
1118 0 : tracing::warn!("weekly history partition {suffix} has no row security: {e}");
1119 0 : Ok(())
1120 : }
1121 : }
1122 2560 : }
1123 :
1124 : /// Recovery for a range DEFAULT already holds rows for, in execution order:
1125 : /// build the week's table STANDALONE (a `PARTITION OF` would fail again), move
1126 : /// exactly `[lo, hi)` out of DEFAULT into it, then ATTACH. The move is what
1127 : /// makes the ATTACH legal, and its bounds are what keep every other row in
1128 : /// DEFAULT. Run as one unit — a partial application would strand rows in an
1129 : /// unattached table.
1130 12 : fn adopt_default_rows_sql(keyed: bool, suffix: &str, lo: &str, hi: &str) -> [String; 3] {
1131 12 : if keyed {
1132 2 : return [
1133 2 : format!(
1134 2 : "CREATE TABLE IF NOT EXISTS attr_rows_{suffix} \
1135 2 : (LIKE attr_rows_long INCLUDING DEFAULTS INCLUDING CONSTRAINTS)"
1136 2 : ),
1137 2 : format!(
1138 2 : "WITH moved AS (DELETE FROM attr_rows_default \
1139 2 : WHERE observed_at >= '{lo}' AND observed_at < '{hi}' RETURNING *) \
1140 2 : INSERT INTO attr_rows_{suffix} SELECT * FROM moved"
1141 2 : ),
1142 2 : format!(
1143 2 : "ALTER TABLE attr_rows_long ATTACH PARTITION attr_rows_{suffix} \
1144 2 : FOR VALUES FROM ('{lo}') TO ('{hi}')"
1145 2 : ),
1146 2 : ];
1147 10 : }
1148 10 : [
1149 10 : format!(
1150 10 : "CREATE TABLE IF NOT EXISTS attr_instances_{suffix} \
1151 10 : (LIKE attr_instances INCLUDING DEFAULTS INCLUDING CONSTRAINTS)"
1152 10 : ),
1153 10 : format!(
1154 10 : "WITH moved AS (DELETE FROM attr_instances_default \
1155 10 : WHERE observed_at >= '{lo}' AND observed_at < '{hi}' RETURNING *) \
1156 10 : INSERT INTO attr_instances_{suffix} SELECT * FROM moved"
1157 10 : ),
1158 10 : format!(
1159 10 : "ALTER TABLE attr_instances ATTACH PARTITION attr_instances_{suffix} \
1160 10 : FOR VALUES FROM ('{lo}') TO ('{hi}')"
1161 10 : ),
1162 10 : ]
1163 12 : }
1164 :
1165 : /// Entities whose stored current state one tick rewrites at most; the rest
1166 : /// wait for the next tick, earliest expiry first.
1167 : const PRUNE_BATCH: i64 = 1000;
1168 :
1169 : /// History rows one reap transaction deletes at most, in one table.
1170 : const REAP_BATCH: i64 = 1000;
1171 :
1172 : /// Batches one history table gets a pass; the rest wait for the next.
1173 : const REAP_BATCHES: usize = 100;
1174 :
1175 : /// The tables that hold history rows, smallest first: every partition or
1176 : /// timescale chunk under `attr_instances` and `attr_rows` (`attr_rows_long`'s
1177 : /// weeks and `attr_rows_short` included), and `attr_versions`. Each comes
1178 : /// quoted and schema-qualified, with its `expires_at` index (migration 0030)
1179 : /// named bare and qualified, and that index's validity, NULL without one.
1180 : /// A hypertable or partitioned parent is no table here: its rows live in its
1181 : /// children. Catalog only, as under forced RLS no count would see a row.
1182 : const HISTORY_TABLES: &str = "WITH RECURSIVE tree(oid) AS (
1183 : SELECT i.inhrelid FROM pg_inherits i
1184 : WHERE i.inhparent IN ('public.attr_instances'::regclass, 'public.attr_rows'::regclass)
1185 : UNION ALL
1186 : SELECT i.inhrelid FROM pg_inherits i JOIN tree t ON i.inhparent = t.oid)
1187 : SELECT format('%I.%I', n.nspname, c.relname),
1188 : quote_ident(left(c.relname, 54) || '_expires'),
1189 : format('%I.%I', n.nspname, left(c.relname, 54) || '_expires'),
1190 : x.indisvalid
1191 : FROM (SELECT oid FROM tree UNION SELECT 'public.attr_versions'::regclass::oid) l
1192 : JOIN pg_class c ON c.oid = l.oid
1193 : JOIN pg_namespace n ON n.oid = c.relnamespace
1194 : LEFT JOIN pg_class ic ON ic.relname = left(c.relname, 54) || '_expires'
1195 : AND ic.relnamespace = c.relnamespace
1196 : LEFT JOIN pg_index x ON x.indexrelid = ic.oid AND x.indrelid = c.oid
1197 : WHERE c.relkind = 'r'
1198 : ORDER BY pg_relation_size(c.oid), 1";
1199 :
1200 : /// One row of [`HISTORY_TABLES`].
1201 : type HistoryTable = (String, String, String, Option<bool>);
1202 :
1203 : /// The 4.22 expired-instance reap: the current state's expired instances
1204 : /// (pruned from `entities.entity`), then the history rows. Isolated so a
1205 : /// deadlock with concurrent ingest never poisons the main maintenance
1206 : /// transaction. Service role: the reap is cross-tenant work (RLS would hide
1207 : /// other tenants' rows).
1208 : ///
1209 : /// The prune is bounded (`PRUNE_BATCH`, by the `attrs_expire_at` index) and
1210 : /// commits first. The history rows go table by table through each table's
1211 : /// `expires_at` index ([`history_expiry_index`]), `REAP_BATCH` rows a
1212 : /// transaction ([`expired_history_sql`]); a table without a valid index yet,
1213 : /// and every row the column misses, wait for [`expiry_scan_daily`].
1214 1007 : async fn reap_expired_instances(pool: &PgPool) -> Result<u64, sqlx::Error> {
1215 1007 : let mut tx = pool.begin().await?;
1216 1007 : crate::store::pg::set_service(&mut tx).await?;
1217 1007 : let mut n = super::entity::prune_expired_current(&mut tx, PRUNE_BATCH).await?;
1218 1007 : tx.commit().await?;
1219 1007 : let tables: Vec<HistoryTable> = sqlx::query_as(HISTORY_TABLES).fetch_all(pool).await?;
1220 18878 : for (table, _, _, valid) in tables {
1221 18878 : if valid != Some(true) {
1222 1 : continue;
1223 18877 : }
1224 18877 : let sql = expired_history_sql(&table);
1225 18877 : for _ in 0..REAP_BATCHES {
1226 18877 : let mut tx = pool.begin().await?;
1227 18877 : crate::store::pg::set_service(&mut tx).await?;
1228 18877 : let gone = sqlx::query(sqlx::AssertSqlSafe(sql.clone()))
1229 18877 : .execute(&mut *tx)
1230 18877 : .await?
1231 18876 : .rows_affected();
1232 18876 : tx.commit().await?;
1233 18876 : n = n.saturating_add(gone);
1234 18876 : if gone < REAP_BATCH as u64 {
1235 18876 : break;
1236 0 : }
1237 : }
1238 : }
1239 1006 : Ok(n)
1240 1007 : }
1241 :
1242 : /// One batch of the 4.22 history reap on `table`, a quoted name from
1243 : /// `HISTORY_TABLES`: at most `REAP_BATCH` rows its `expires_at` index finds
1244 : /// past, deleted by ctid. `ONLY` the table, as a ctid names a row of one
1245 : /// table. The row's `data` decides: a correction by an older release leaves
1246 : /// the column stale (0030's rollback note), and a row whose `expiresAt` has
1247 : /// not passed is kept. try_timestamptz runs on the batch's rows alone.
1248 18885 : pub fn expired_history_sql(table: &str) -> String {
1249 18885 : format!(
1250 : "DELETE FROM ONLY {table} WHERE ctid = ANY(ARRAY(
1251 : SELECT ctid FROM ONLY {table} WHERE expires_at < now() LIMIT {REAP_BATCH}))
1252 : AND expires_at < now() AND try_timestamptz(data->>'expiresAt') < now()"
1253 : )
1254 18885 : }
1255 :
1256 : /// The history rows `expires_at` misses, by a scan of `data`: rows written
1257 : /// before migration 0030 or by an older release after a rollback, the tables
1258 : /// [`history_expiry_index`] has not reached, and a timescale chunk's
1259 : /// compressed rows, which no index of the chunk holds. Every row of both
1260 : /// parents and the versions is read, so the scan runs once a day, claimed on
1261 : /// its own `maintenance_jobs` row and held there until done; a statement
1262 : /// that fails (statement_timeout on a large history) costs only itself, and
1263 : /// the day is spent either way, as a retry every tick would be the cost this
1264 : /// replaces. `data ? 'expiresAt'` keeps the plpgsql cast and its
1265 : /// subtransaction to the rows that carry one. `None` when not due.
1266 1007 : async fn expiry_scan_daily(pool: &PgPool) -> Result<Option<u64>, sqlx::Error> {
1267 1007 : let mut tx = pool.begin().await?;
1268 1007 : let due = sqlx::query(
1269 1007 : "SELECT name FROM maintenance_jobs WHERE name = 'history_expiry_scan'
1270 1007 : AND (last_run IS NULL OR last_run < now() - interval '1 day')
1271 1007 : FOR UPDATE SKIP LOCKED",
1272 1007 : )
1273 1007 : .fetch_optional(&mut *tx)
1274 1007 : .await?;
1275 1007 : if due.is_none() {
1276 1005 : return Ok(None);
1277 2 : }
1278 2 : crate::store::pg::set_service(&mut tx).await?;
1279 2 : let (mut n, mut failed) = (0u64, None);
1280 : // try_timestamptz (0001_init.sql), not a bare cast: `expiresAt` is
1281 : // jsonb TEXT and a stamp PostgreSQL cannot parse would abort this
1282 : // DELETE for the whole deployment, every day, forever.
1283 6 : for sql in [
1284 2 : "DELETE FROM attr_instances WHERE data ? 'expiresAt' AND try_timestamptz(data->>'expiresAt') < now()",
1285 2 : "DELETE FROM attr_rows WHERE data ? 'expiresAt' AND try_timestamptz(data->>'expiresAt') < now()",
1286 2 : "DELETE FROM attr_versions WHERE data ? 'expiresAt' AND try_timestamptz(data->>'expiresAt') < now()",
1287 2 : ] {
1288 6 : let mut sp = tx.begin().await?;
1289 6 : match sqlx::query(sql).execute(&mut *sp).await {
1290 6 : Ok(r) => {
1291 6 : sp.commit().await?;
1292 6 : n = n.saturating_add(r.rows_affected());
1293 : }
1294 0 : Err(e) => {
1295 0 : sp.rollback().await?;
1296 0 : failed = Some(e);
1297 : }
1298 : }
1299 : }
1300 2 : sqlx::query("UPDATE maintenance_jobs SET last_run = now() WHERE name = 'history_expiry_scan'")
1301 2 : .execute(&mut *tx)
1302 2 : .await?;
1303 2 : tx.commit().await?;
1304 2 : match failed {
1305 0 : Some(e) => Err(e),
1306 2 : None => Ok(Some(n)),
1307 : }
1308 1007 : }
1309 :
1310 : /// How long a pass of [`history_expiry_index`] starts new builds for.
1311 : const EXPIRY_INDEX_BUDGET: std::time::Duration = std::time::Duration::from_secs(10);
1312 :
1313 : /// The partial index on `expires_at` (migration 0030) that
1314 : /// `reap_expired_instances` reads, built on each history table of
1315 : /// `HISTORY_TABLES` by name with `CREATE INDEX CONCURRENTLY` (writes go
1316 : /// on), smallest first, as [`history_one_index`] builds 0017's: never by a
1317 : /// migration, never on a parent. A pass starts builds while it has spent
1318 : /// less than `EXPIRY_INDEX_BUDGET`, so the empty weeks created ahead are
1319 : /// indexed in the pass that creates them, and each large table of a history
1320 : /// written before the migration takes a pass of its own. A timescale chunk
1321 : /// is a table as well: an index on the hypertable would build every chunk
1322 : /// at once under a lock that blocks writes. `None` when every table has
1323 : /// one, or while another pod builds.
1324 1010 : pub async fn history_expiry_index(pool: &PgPool) -> Result<Option<String>, sqlx::Error> {
1325 1010 : let missing: Vec<HistoryTable> = sqlx::query_as::<_, HistoryTable>(HISTORY_TABLES)
1326 1010 : .fetch_all(pool)
1327 1010 : .await?
1328 1010 : .into_iter()
1329 18915 : .filter(|t| t.3 != Some(true))
1330 1010 : .collect();
1331 1010 : if missing.is_empty() {
1332 993 : return Ok(None);
1333 17 : }
1334 : // a concurrent build runs outside a transaction, on a session of its own
1335 : // the pool never sees again (as in `history_one_index`)
1336 17 : let mut conn = pool.acquire().await?.detach();
1337 17 : let held: bool =
1338 17 : sqlx::query_scalar("SELECT pg_try_advisory_lock(hashtext('antares.history_expiry_index'))")
1339 17 : .fetch_one(&mut conn)
1340 17 : .await?;
1341 17 : if !held {
1342 0 : return Ok(None);
1343 17 : }
1344 17 : let built = expiry_index_steps(&mut conn, &missing).await;
1345 : use sqlx::Connection;
1346 : // Closing the session releases the advisory lock with it.
1347 17 : conn.close().await?;
1348 17 : let built = built?;
1349 17 : Ok(Some(format!(
1350 17 : "history expiry index built on {built} of {} tables (4.22)",
1351 17 : missing.len()
1352 17 : )))
1353 1010 : }
1354 :
1355 17 : async fn expiry_index_steps(
1356 17 : conn: &mut sqlx::PgConnection,
1357 17 : missing: &[HistoryTable],
1358 17 : ) -> Result<usize, sqlx::Error> {
1359 : use sqlx::Executor;
1360 : // A concurrent build waits for older transactions to end: those waits
1361 : // count against lock_timeout, and one that runs out leaves an invalid
1362 : // index the next pass drops and builds again. Short, unlike the 0017
1363 : // build's: the reaps of this tick run after it, and a long transaction
1364 : // open somewhere would hold them for the whole wait.
1365 17 : conn.execute("SET statement_timeout = 0; SET lock_timeout = '5s'")
1366 17 : .await?;
1367 17 : let started = std::time::Instant::now();
1368 17 : let mut built: usize = 0;
1369 34 : for (table, bare, index, valid) in missing {
1370 34 : if started.elapsed() >= EXPIRY_INDEX_BUDGET {
1371 0 : break;
1372 34 : }
1373 34 : if valid.is_some() {
1374 0 : conn.execute(sqlx::AssertSqlSafe(format!(
1375 0 : "DROP INDEX CONCURRENTLY IF EXISTS {index}"
1376 0 : )))
1377 0 : .await?;
1378 34 : }
1379 34 : conn.execute(sqlx::AssertSqlSafe(format!(
1380 34 : "CREATE INDEX CONCURRENTLY IF NOT EXISTS {bare} ON {table} (expires_at) \
1381 34 : WHERE expires_at IS NOT NULL"
1382 34 : )))
1383 34 : .await?;
1384 34 : built = built.saturating_add(1);
1385 : }
1386 17 : Ok(built)
1387 17 : }
1388 :
1389 : /// The columns of the one history btree (migration 0017).
1390 : const ONE_INDEX_COLUMNS: &str = "(tenant_id, entity_id, attr_id, observed_at, instance_id)";
1391 :
1392 : /// The standalone index [`history_one_index`] builds on partition `part`,
1393 : /// quoted bare (CREATE INDEX puts it in the table's schema) and qualified,
1394 : /// and the partition qualified. A weekly suffix keeps the name under the
1395 : /// 63-byte identifier limit.
1396 12 : fn one_index_names(part: &str) -> (String, String, String) {
1397 24 : let q = |s: &str| format!("\"{}\"", s.replace('"', "\"\""));
1398 12 : let index = q(&format!("{part}_u_time"));
1399 12 : (
1400 12 : index.clone(),
1401 12 : format!("public.{index}"),
1402 12 : format!("public.{}", q(part)),
1403 12 : )
1404 12 : }
1405 :
1406 : /// The partitions of `attr_instances` without a valid `{partition}_u_time`
1407 : /// index, with whether an invalid one (a build that failed) is in the way.
1408 : const PARTITIONS_WITHOUT_ONE_INDEX: &str = "SELECT c.relname, x.indexrelid IS NOT NULL AS stale
1409 : FROM pg_inherits i
1410 : JOIN pg_class c ON c.oid = i.inhrelid
1411 : LEFT JOIN pg_class ic ON ic.relname = c.relname || '_u_time'
1412 : AND ic.relnamespace = 'public'::regnamespace
1413 : LEFT JOIN pg_index x ON x.indexrelid = ic.oid
1414 : WHERE i.inhparent = 'public.attr_instances'::regclass
1415 : AND x.indisvalid IS NOT TRUE
1416 : ORDER BY pg_relation_size(c.oid)";
1417 :
1418 : /// Move a plain-mode history onto one btree (migration 0017), one
1419 : /// step a pass: while `i_attr_instances_lookup` exists, build
1420 : /// `{partition}_u_time` on the smallest partition lacking it with `CREATE
1421 : /// INDEX CONCURRENTLY` (writes go on), and once every partition has its own,
1422 : /// make them the partitions of `u_attr_instances_time` and drop the two old
1423 : /// indexes in one transaction of catalog changes held to a 2 s
1424 : /// `lock_timeout`. Standalone indexes first, never an invalid parent filled
1425 : /// in place: a partition retention drops before its turn would leave such a
1426 : /// parent invalid for good, as PostgreSQL validates it only on an attach.
1427 : /// `None` once done, or while another pod holds the step.
1428 : ///
1429 : /// The build takes the disk of one more index per partition until
1430 : /// the swap frees two; a timescale hypertable keeps both (no concurrent
1431 : /// unique build there).
1432 521 : pub async fn history_one_index(pool: &PgPool) -> Result<Option<String>, sqlx::Error> {
1433 521 : let pending: bool =
1434 521 : sqlx::query_scalar("SELECT to_regclass('public.i_attr_instances_lookup') IS NOT NULL")
1435 521 : .fetch_one(pool)
1436 521 : .await?;
1437 521 : if !pending {
1438 514 : return Ok(None);
1439 7 : }
1440 : // CREATE INDEX CONCURRENTLY runs outside a transaction and for as long as
1441 : // the partition takes, so it gets a connection of its own the pool never
1442 : // sees again (as the migrations do in `connect_with`).
1443 7 : let mut conn = pool.acquire().await?.detach();
1444 7 : let held: bool =
1445 7 : sqlx::query_scalar("SELECT pg_try_advisory_lock(hashtext('antares.history_one_index'))")
1446 7 : .fetch_one(&mut conn)
1447 7 : .await?;
1448 7 : if !held {
1449 0 : return Ok(None);
1450 7 : }
1451 7 : let line = one_index_step(&mut conn).await;
1452 : use sqlx::Connection;
1453 : // Closing the session releases the advisory lock with it.
1454 7 : conn.close().await?;
1455 7 : line.map(Some)
1456 521 : }
1457 :
1458 7 : async fn one_index_step(conn: &mut sqlx::PgConnection) -> Result<String, sqlx::Error> {
1459 : use sqlx::Executor;
1460 7 : let missing: Vec<(String, bool)> = sqlx::query_as(PARTITIONS_WITHOUT_ONE_INDEX)
1461 7 : .fetch_all(&mut *conn)
1462 7 : .await?;
1463 7 : if let Some((part, stale)) = missing.first() {
1464 6 : let (bare, index, table) = one_index_names(part);
1465 : // A concurrent build waits for older transactions to end: those waits
1466 : // count against lock_timeout, and one that runs out leaves an invalid
1467 : // index the next pass drops and builds again.
1468 6 : conn.execute("SET statement_timeout = 0; SET lock_timeout = '60s'")
1469 6 : .await?;
1470 6 : if *stale {
1471 0 : conn.execute(sqlx::AssertSqlSafe(format!(
1472 0 : "DROP INDEX CONCURRENTLY IF EXISTS {index}"
1473 0 : )))
1474 0 : .await?;
1475 6 : }
1476 6 : conn.execute(sqlx::AssertSqlSafe(format!(
1477 6 : "CREATE UNIQUE INDEX CONCURRENTLY {bare} ON {table} {ONE_INDEX_COLUMNS}"
1478 6 : )))
1479 6 : .await?;
1480 6 : return Ok(format!("history index built on {part}"));
1481 1 : }
1482 1 : let mut tx = sqlx::Connection::begin(&mut *conn).await?;
1483 1 : sqlx::query("SET LOCAL lock_timeout = '2s'")
1484 1 : .execute(&mut *tx)
1485 1 : .await?;
1486 : // The partition set is read again under the parent's lock: a week created
1487 : // since has no index of its own yet, and its turn comes next pass.
1488 1 : sqlx::query("LOCK TABLE public.attr_instances IN SHARE MODE")
1489 1 : .execute(&mut *tx)
1490 1 : .await?;
1491 1 : let late: Option<String> = sqlx::query_scalar(PARTITIONS_WITHOUT_ONE_INDEX)
1492 1 : .fetch_optional(&mut *tx)
1493 1 : .await?;
1494 1 : if late.is_some() {
1495 0 : tx.rollback().await?;
1496 0 : return Ok("history index: a new partition waits for its own".into());
1497 1 : }
1498 1 : let parts: Vec<String> = sqlx::query_scalar(
1499 1 : "SELECT c.relname FROM pg_inherits i JOIN pg_class c ON c.oid = i.inhrelid
1500 1 : WHERE i.inhparent = 'public.attr_instances'::regclass",
1501 1 : )
1502 1 : .fetch_all(&mut *tx)
1503 1 : .await?;
1504 1 : sqlx::query(sqlx::AssertSqlSafe(format!(
1505 1 : "CREATE UNIQUE INDEX u_attr_instances_time ON ONLY public.attr_instances {ONE_INDEX_COLUMNS}"
1506 1 : )))
1507 1 : .execute(&mut *tx)
1508 1 : .await?;
1509 6 : for part in &parts {
1510 6 : let (_, index, _) = one_index_names(part);
1511 6 : sqlx::query(sqlx::AssertSqlSafe(format!(
1512 6 : "ALTER INDEX public.u_attr_instances_time ATTACH PARTITION {index}"
1513 6 : )))
1514 6 : .execute(&mut *tx)
1515 6 : .await?;
1516 : }
1517 1 : sqlx::query("DROP INDEX public.i_attr_instances_lookup, public.u_attr_instances")
1518 1 : .execute(&mut *tx)
1519 1 : .await?;
1520 1 : tx.commit().await?;
1521 1 : Ok(format!(
1522 1 : "history on one index: {} partitions swapped",
1523 1 : parts.len()
1524 1 : ))
1525 7 : }
1526 :
1527 : #[cfg(test)]
1528 : mod tests {
1529 : use super::*;
1530 :
1531 : const LO: &str = "2026-08-17T00:00:00+00";
1532 : const HI: &str = "2026-08-24T00:00:00+00";
1533 :
1534 : /// The partition name is an identifier the catalog hands back, not SQL.
1535 : /// A bare interpolation drops any partition whose name needs quoting and
1536 : /// resolves through `search_path` instead of the schema the row is in.
1537 : #[test]
1538 2 : fn a_partition_is_dropped_by_the_name_the_catalog_spells() {
1539 2 : assert_eq!(
1540 2 : drop_partition_sql("attr_instances_2026w34"),
1541 : "DROP TABLE public.\"attr_instances_2026w34\""
1542 : );
1543 2 : assert_eq!(
1544 2 : drop_partition_sql("Weird Name"),
1545 : "DROP TABLE public.\"Weird Name\"",
1546 : "a name that is not a bare identifier still names one table"
1547 : );
1548 2 : assert_eq!(
1549 2 : drop_partition_sql("a\"b"),
1550 : "DROP TABLE public.\"a\"\"b\"",
1551 : "a quote in the name closes nothing"
1552 : );
1553 2 : }
1554 :
1555 : #[test]
1556 2 : fn create_partition_covers_exactly_the_week() {
1557 2 : let sql = create_partition_sql(false, "2026w34", LO, HI);
1558 2 : assert!(sql.contains("attr_instances_2026w34 PARTITION OF attr_instances"));
1559 2 : assert!(sql.contains(&format!("FROM ('{LO}') TO ('{HI}')")));
1560 2 : }
1561 :
1562 : /// The week's table must be built standalone: `CREATE ... PARTITION OF`
1563 : /// is the statement that just failed, so repeating it cannot recover the
1564 : /// range.
1565 : #[test]
1566 2 : fn adopt_builds_the_table_standalone_then_attaches_it() {
1567 2 : let [create, _move, attach] = adopt_default_rows_sql(false, "2026w34", LO, HI);
1568 2 : assert!(create.contains("LIKE attr_instances"), "{create}");
1569 2 : assert!(!create.contains("PARTITION OF"), "{create}");
1570 2 : assert!(
1571 2 : attach
1572 2 : .starts_with("ALTER TABLE attr_instances ATTACH PARTITION attr_instances_2026w34"),
1573 : "{attach}"
1574 : );
1575 2 : assert!(
1576 2 : attach.contains(&format!("FROM ('{LO}') TO ('{HI}')")),
1577 : "{attach}"
1578 : );
1579 2 : }
1580 :
1581 : /// The move is bounded by the partition range on BOTH sides. Unbounded, it
1582 : /// would empty the DEFAULT partition of every historic-backfill row in the
1583 : /// deployment and stuff them into one week.
1584 : #[test]
1585 2 : fn adopt_moves_only_the_partition_range_out_of_default() {
1586 2 : let [_create, mv, _attach] = adopt_default_rows_sql(false, "2026w34", LO, HI);
1587 2 : assert!(mv.contains("DELETE FROM attr_instances_default"), "{mv}");
1588 2 : assert!(mv.contains(&format!("observed_at >= '{LO}'")), "{mv}");
1589 2 : assert!(mv.contains(&format!("observed_at < '{HI}'")), "{mv}");
1590 2 : assert!(
1591 2 : mv.contains("INSERT INTO attr_instances_2026w34"),
1592 : "moved rows must land in the week's table: {mv}"
1593 : );
1594 2 : }
1595 :
1596 : /// A week from the cut-over on is a week of `attr_rows`, built,
1597 : /// moved and attached under its logged child `attr_rows_long` (0023)
1598 : /// only.
1599 : #[test]
1600 2 : fn a_keyed_week_goes_to_attr_rows() {
1601 2 : let sql = create_partition_sql(true, "2026w34", LO, HI);
1602 2 : assert!(
1603 2 : sql.contains("attr_rows_2026w34 PARTITION OF attr_rows_long "),
1604 : "{sql}"
1605 : );
1606 2 : let [create, mv, attach] = adopt_default_rows_sql(true, "2026w34", LO, HI);
1607 2 : assert!(create.contains("(LIKE attr_rows_long "), "{create}");
1608 2 : assert!(mv.contains("DELETE FROM attr_rows_default"), "{mv}");
1609 2 : assert!(mv.contains("INSERT INTO attr_rows_2026w34"), "{mv}");
1610 2 : assert!(mv.contains(&format!("observed_at < '{HI}'")), "{mv}");
1611 2 : assert!(
1612 2 : attach.starts_with("ALTER TABLE attr_rows_long ATTACH PARTITION attr_rows_2026w34"),
1613 : "{attach}"
1614 : );
1615 2 : assert!(!format!("{sql}{create}{mv}{attach}").contains("attr_instances"));
1616 2 : }
1617 : }
|
