autumn_web/counter_cache.rs
1//! Declarative counter caches (issue #1325).
2//!
3//! A counter cache is a denormalised `{child}_count` column on a parent row
4//! that the framework keeps current: `posts.comment_count`,
5//! `subreddits.subscriber_count`, `teams.member_count`. Declaring it on the
6//! child's association —
7//!
8//! ```rust,ignore
9//! #[autumn_web::model]
10//! #[belongs_to(Post, counter_cache)]
11//! pub struct Comment { /* … */ }
12//! ```
13//!
14//! — makes the generated `#[repository]` maintain `posts.comment_count` on
15//! every insert, delete, soft delete, restore and foreign-key reassignment,
16//! **inside the same transaction as the row mutation**, with a single atomic
17//! `UPDATE posts SET comment_count = comment_count + $1` (never a
18//! read-modify-write, so concurrent inserts cannot lose updates).
19//!
20//! # Why the maintenance lives on the child
21//!
22//! The `#[model]` and `#[repository]` derives are separate proc-macro
23//! invocations, so the repository macro never sees the model struct's
24//! `#[belongs_to]` attributes — exactly the problem
25//! [`crate::repository::AutumnDependents`] already solves for dependent
26//! cascades. [`AutumnCounterCaches`] is the same bridge: `#[model]` emits an
27//! *inherent* `counter_caches()` that shadows the empty blanket, and the
28//! generated repository calls `Model::counter_caches()` by concrete path.
29//!
30//! Because the shadow is inherent, it is **not** visible through a generic
31//! `M: AutumnCounterCaches` bound (a generic call would resolve to the blanket
32//! default). Every helper in this module therefore takes the spec slice as an
33//! argument rather than recovering it from a bound — call them as
34//! `counter_cache_after_insert(conn, Comment::counter_caches(), &record)`.
35//!
36//! # Drift and repair
37//!
38//! Counters are not clamped: a drifted counter can go negative, and that is
39//! deliberate — a negative count is a visible signal, where a silent `GREATEST(0,
40//! …)` would hide the bug. [`counter_cache_recompute`] rebuilds the column from
41//! the source of truth and is idempotent; the generated repository exposes it as
42//! `recompute_counter_caches()` / `recompute_counter_caches_for(parent_id)`.
43//!
44//! # Identifier safety
45//!
46//! Every table/column name in the SQL below arrives as a `&'static str` emitted
47//! by `#[model]`, and the only user-controlled one (`counter_cache = "…"`) is
48//! validated at macro time to be a plain identifier. [`is_plain_identifier`]
49//! re-checks that at run time under `debug_assertions` so a hand-constructed
50//! spec cannot smuggle SQL through `format!`.
51//!
52//! On top of that, every interpolated identifier is [quoted](quote_ident) — the
53//! same convention the `dependent(restrict)` codegen already follows. That is
54//! what lets a counter column be named after a SQL keyword
55//! (`counter_cache = "order"`) without turning every generated statement into a
56//! syntax error, and it is safe precisely because the validation above rules out
57//! the `"` that would otherwise escape the quotes.
58
59use std::collections::HashMap;
60use std::fmt::Write as _;
61
62use diesel::sql_types::{BigInt, Nullable, Text};
63use diesel_async::RunQueryDsl as _;
64use scoped_futures::ScopedFutureExt as _;
65
66use crate::db::{RuntimeConnection, scoped_immediate_transaction};
67use crate::{AutumnError, AutumnResult};
68
69// Backend-forked placeholders: Postgres numbers its binds, `SQLite` does not.
70// Binds are pushed in the same order on both, so one statement template with
71// swapped placeholder text serves both backends.
72#[cfg(not(feature = "sqlite"))]
73const PH1: &str = "$1";
74#[cfg(not(feature = "sqlite"))]
75const PH2: &str = "$2";
76#[cfg(not(feature = "sqlite"))]
77const PH3: &str = "$3";
78#[cfg(feature = "sqlite")]
79const PH1: &str = "?";
80#[cfg(feature = "sqlite")]
81const PH2: &str = "?";
82#[cfg(feature = "sqlite")]
83const PH3: &str = "?";
84
85/// Row lock appended to the sub-selects that read a child row's foreign key.
86///
87/// Without it there is a window between "read the child's current parent" and
88/// "write the child's new parent" in which a concurrent writer can re-parent the
89/// row, so the counter would be moved off the wrong parent. Taking the lock on
90/// the child row closes it: the concurrent writer blocks until this transaction
91/// commits. The lock is on the same row the surrounding mutation locks anyway,
92/// so it introduces no new lock-ordering edge.
93///
94/// `SQLite` has no `SELECT … FOR UPDATE` and needs none: generated write paths
95/// begin with `BEGIN IMMEDIATE`, which excludes every other writer for the
96/// duration — the same reason `maybe_for_update!` is the identity there.
97#[cfg(not(feature = "sqlite"))]
98const FOR_UPDATE: &str = " FOR UPDATE";
99#[cfg(feature = "sqlite")]
100const FOR_UPDATE: &str = "";
101
102/// NULL-safe inequality. Postgres spells it `IS DISTINCT FROM`; `SQLite` has
103/// spelled the same operator `IS NOT` since long before it gained the SQL
104/// standard alias, so the short form is the portable one there.
105#[cfg(not(feature = "sqlite"))]
106const IS_DISTINCT_FROM: &str = "IS DISTINCT FROM";
107#[cfg(feature = "sqlite")]
108const IS_DISTINCT_FROM: &str = "IS NOT";
109
110/// Its negation, for matching a tenant discriminator that may be NULL. Plain `=`
111/// would make two rows that are both untenanted fail to match, which for a
112/// counter cache means the maintenance silently does nothing — the one failure
113/// mode this module works hardest to avoid.
114#[cfg(not(feature = "sqlite"))]
115const IS_NOT_DISTINCT_FROM: &str = "IS NOT DISTINCT FROM";
116#[cfg(feature = "sqlite")]
117const IS_NOT_DISTINCT_FROM: &str = "IS";
118
119/// The alias the generated SQL gives the *child* table whenever it appears in a
120/// sub-select. Always aliasing keeps a **self-referential** counter cache
121/// (a `Comment` that `belongs_to` a parent `Comment` maintaining `reply_count`)
122/// unambiguous, where a bare table name would bind to the outer `UPDATE` target.
123const CHILD_ALIAS: &str = "__autumn_cc_child";
124
125/// The soft-delete marker column. Autumn's `soft_delete` convention is fixed, so
126/// the predicate is a constant rather than another spec field.
127const DELETED_AT: &str = "deleted_at";
128
129/// One counter-cached `belongs_to` leg, produced at compile time by `#[model]`
130/// and consulted at run time by the child's generated repository.
131///
132/// Framework plumbing; not constructed by hand.
133#[derive(Debug)]
134pub struct CounterCacheSpec<M: 'static> {
135 /// The child's table, e.g. `comments`.
136 pub child_table: &'static str,
137 /// The child's primary-key column, e.g. `id`.
138 pub child_pk: &'static str,
139 /// Whether the child model carries a `deleted_at` column. When true the
140 /// count reflects **live rows only**: the decrement is scoped to a live
141 /// child (so a second soft delete moves nothing) and the recompute filters
142 /// soft-deleted rows out.
143 pub child_soft_delete: bool,
144 /// The child's foreign-key column pointing at the parent, e.g. `post_id`.
145 pub fk_column: &'static str,
146 /// The parent's table, e.g. `posts`.
147 pub parent_table: &'static str,
148 /// The parent's primary-key column, e.g. `id`.
149 pub parent_pk: &'static str,
150 /// The maintained column on the parent, e.g. `comment_count`.
151 pub counter_column: &'static str,
152 /// Reads this leg's foreign key off a child record. `None` for a nullable
153 /// foreign key that is unset — an unparented child moves no counter.
154 pub fk_of: fn(&M) -> Option<i64>,
155 /// Reads the child's own primary key off a record. Used by the bulk paths
156 /// to match a post-update record back to the foreign keys captured for it
157 /// before the update, and to scope the tenant predicate below.
158 pub pk_of: fn(&M) -> i64,
159 /// Whether a record is live (not soft-deleted). Always `true` for a child
160 /// with no `deleted_at` column.
161 ///
162 /// The generated `update` does **not** filter soft-deleted rows, so an
163 /// already-deleted child can be re-parented. It is counted by nobody, so
164 /// neither its old nor its new parent may move — without this the update
165 /// would decrement a parent that had already dropped it and increment one
166 /// for a dead row.
167 pub live_of: fn(&M) -> bool,
168 /// The tenant-discriminator column, from
169 /// `#[belongs_to(…, counter_cache, counter_cache_tenant = "<column>")]`.
170 ///
171 /// A counter update is a write to a row the caller only had to name the id
172 /// of, so on a shared multi-tenant table it has to be confined to the
173 /// caller's own tenant — the same hazard `#[votable]`'s aggregate `UPDATE`
174 /// is scoped for. The predicate spells the invariant "the parent sits in the
175 /// same tenant as the child whose foreign key named it", so **both** tables
176 /// must carry a column of this name. That is why it is explicit: `#[model]`
177 /// on the child cannot see the parent's fields, and guessing would turn
178 /// every tenant-scoped child hanging off a global parent into a hard
179 /// `column does not exist`.
180 ///
181 /// `None` (the default) emits no predicate anywhere, so a single-tenant
182 /// app's SQL is byte-for-byte what it would be without this field.
183 pub tenant_column: Option<&'static str>,
184 /// Reads the child's tenant discriminator off a record, as text, the way
185 /// the database spells `CAST(<tenant> AS TEXT)`.
186 ///
187 /// `Some` only for a leg with a [`Self::tenant_column`] whose child model
188 /// carries that field as an integer or string. Then the pre-update capture
189 /// reads the old tenant too, and a child moved to another tenant takes its
190 /// contribution off the old parent under the *old* tenant: the live
191 /// predicate would look for the parent under the new one and miss it.
192 /// `None` keeps the leg's SQL exactly as it was.
193 pub tenant_of: Option<fn(&M) -> Option<String>>,
194 /// This row's weight in the maintained aggregate. `0` excludes the row.
195 ///
196 /// A plain counter cache returns `1` for every row; a `#[derivation]`
197 /// returns `0` when its filter rejects the row, and otherwise `1` (a count)
198 /// or the summed field's value (a sum).
199 pub contrib_of: fn(&M) -> i64,
200 /// The SQL half of [`Self::contrib_of`], for the set-based paths.
201 ///
202 /// `"1"` for a counter cache, else a child column reference such as
203 /// `{c}."score"`. `{c}` is the placeholder for whichever alias the statement
204 /// gives the child table.
205 pub contrib_sql: &'static str,
206 /// The derivation's row filter, lowered to SQL and already prefixed with
207 /// ` AND ` so any builder can concatenate it.
208 ///
209 /// `""` for a counter cache, which is what keeps a counter cache's
210 /// generated SQL byte-identical. Else ` AND (<pred>)`, using the same `{c}`
211 /// child-alias placeholder as [`Self::contrib_sql`].
212 pub filter_sql: &'static str,
213 /// The `#[derivation]` this spec maintains, or `None` for a plain counter
214 /// cache. It carries the metadata the backfill and status surfaces read.
215 pub derivation: Option<&'static crate::derivation::DerivationDef>,
216}
217
218// A manual `Clone`/`Copy` (rather than a derive) because the derive would add a
219// `M: Clone` bound the model types need not satisfy.
220impl<M: 'static> Clone for CounterCacheSpec<M> {
221 fn clone(&self) -> Self {
222 *self
223 }
224}
225impl<M: 'static> Copy for CounterCacheSpec<M> {}
226
227/// Exposes a model's counter-cache specs to its generated repository (#1325).
228///
229/// A blanket impl returns an empty slice (and `HAS_COUNTER_CACHES = false`) for
230/// every type; `#[model]` emits **inherent** items that shadow both when the
231/// model declares at least one `#[belongs_to(…, counter_cache)]`. This mirrors
232/// [`crate::repository::AutumnDependents`], and carries the same caveat: the
233/// shadow is only visible through a *concrete* path (`Comment::counter_caches()`),
234/// which is exactly how the generated code names it.
235pub trait AutumnCounterCaches: Sized + 'static {
236 /// Whether this model has any counter cache at all.
237 ///
238 /// The generated repository branches on this to decide whether a mutation
239 /// path that would otherwise run without a transaction needs to open one.
240 /// It is a `const`, so for the overwhelmingly common `false` case the whole
241 /// counter-cache path is dead code the optimizer drops.
242 const HAS_COUNTER_CACHES: bool = false;
243
244 /// The model's counter-cache specs, in declaration order.
245 #[must_use]
246 fn counter_caches() -> &'static [CounterCacheSpec<Self>] {
247 &[]
248 }
249}
250
251// Blanket fallback — any type without inherent counter-cache items (i.e. a model
252// with no counter-cached association, or a type not built through `#[model]`).
253impl<T: Sized + 'static> AutumnCounterCaches for T {}
254
255/// Whether `s` is a plain SQL/Rust identifier: ASCII alphanumerics and
256/// underscores, not starting with a digit, non-empty.
257///
258/// `pub` (inside this `pub(crate)` module, so still crate-private) because
259/// [`crate::commentable`] (#1367) builds its polymorphic
260/// statements to the same identifier-safety contract and must apply the same
261/// backstop — one definition, so the two can never drift apart.
262///
263/// Names reaching the `format!`ed SQL below are macro-emitted and already
264/// validated at macro time; this is the run-time backstop for that invariant.
265#[must_use]
266pub fn is_plain_identifier(s: &str) -> bool {
267 !s.is_empty()
268 && !s.starts_with(|c: char| c.is_ascii_digit())
269 && s.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
270}
271
272/// Guard that every identifier a view splices into SQL is plain.
273///
274/// A hard assert, not a debug one. The module doc claims a hand-built spec
275/// cannot smuggle SQL, so a release build has to honour that claim too.
276///
277/// `filter_sql` is deliberately not scanned for SQL syntax. It is a lowered
278/// fragment of quoted identifiers, literals, operators and the `{c}` alias
279/// placeholder, so a substring scan would reject a legal quoted literal:
280/// `filter = status == "a -- b"` carries a comment marker inside a string, and
281/// the macro already escapes and validates the literal it lowers. What matters
282/// at run time is that no placeholder survives substitution, which
283/// [`with_alias`] asserts.
284///
285/// `contrib_sql` is checked structurally instead of by substring: the lowering
286/// only ever produces the literal `1` or one quoted child column.
287fn assert_spec_idents(view: &SqlView) {
288 assert!(
289 is_plain_identifier(view.child_table)
290 && is_plain_identifier(view.child_pk)
291 && is_plain_identifier(view.fk_column)
292 && is_plain_identifier(view.parent_table)
293 && is_plain_identifier(view.parent_pk)
294 && is_plain_identifier(view.counter_column)
295 && view.tenant_column.is_none_or(is_plain_identifier),
296 "counter-cache spec carries a non-identifier name; it would be spliced \
297 verbatim into SQL"
298 );
299 assert!(
300 is_lowered_contribution(view.contrib_sql),
301 "counter-cache spec carries a contribution that is neither `1` nor one \
302 child column; it would be spliced verbatim into SQL"
303 );
304}
305
306/// Whether `sql` is a contribution expression the `#[derivation]` lowering can
307/// produce: the literal `1` (a row count), or one plain child column behind the
308/// `{c}` alias placeholder (a weighted sum).
309fn is_lowered_contribution(sql: &str) -> bool {
310 if sql == "1" {
311 return true;
312 }
313 sql.strip_prefix("{c}.\"")
314 .and_then(|rest| rest.strip_suffix('"'))
315 .is_some_and(is_plain_identifier)
316}
317
318/// Wrap `ident` in the SQL identifier quotes both backends accept.
319///
320/// Quoting is what keeps a counter column named after a SQL keyword
321/// (`counter_cache = "order"`) from turning every generated statement into a
322/// syntax error, and it matches how the `dependent(restrict)` codegen already
323/// interpolates table and column names. Safe to apply unconditionally: the names
324/// are validated plain identifiers, so none of them contains the `"` that would
325/// otherwise let a hand-built spec escape the quotes.
326pub fn quote_ident(ident: &str) -> String {
327 format!("\"{ident}\"")
328}
329
330/// Everything the statement builders need from a spec, with no model type.
331///
332/// [`CounterCacheSpec`] is generic over the child model, but no SQL in this
333/// module depends on that type, and the derivation repair paths
334/// ([`crate::derivation`]) have only a [`crate::derivation::DerivationDef`],
335/// never an `M`. Both therefore build their statements from this view, so one
336/// set of builders serves both and the two can never emit different SQL.
337#[derive(Debug, Clone, Copy)]
338pub struct SqlView {
339 pub child_table: &'static str,
340 pub child_pk: &'static str,
341 pub child_soft_delete: bool,
342 pub fk_column: &'static str,
343 pub parent_table: &'static str,
344 pub parent_pk: &'static str,
345 pub counter_column: &'static str,
346 pub contrib_sql: &'static str,
347 pub filter_sql: &'static str,
348 pub tenant_column: Option<&'static str>,
349}
350
351impl SqlView {
352 /// Whether this is a plain counter cache: every live row contributes 1 and
353 /// nothing is filtered out. The builders below keep such a view's SQL
354 /// byte-identical to what it was before derivations existed.
355 const fn is_plain(&self) -> bool {
356 self.contrib_sql.len() == 1
357 && self.contrib_sql.as_bytes()[0] == b'1'
358 && self.filter_sql.is_empty()
359 }
360
361 /// Whether the aggregate is a row count rather than a weighted sum.
362 const fn counts_rows(&self) -> bool {
363 self.contrib_sql.len() == 1 && self.contrib_sql.as_bytes()[0] == b'1'
364 }
365}
366
367const fn view<M: 'static>(spec: &CounterCacheSpec<M>) -> SqlView {
368 SqlView {
369 child_table: spec.child_table,
370 child_pk: spec.child_pk,
371 child_soft_delete: spec.child_soft_delete,
372 fk_column: spec.fk_column,
373 parent_table: spec.parent_table,
374 parent_pk: spec.parent_pk,
375 counter_column: spec.counter_column,
376 contrib_sql: spec.contrib_sql,
377 filter_sql: spec.filter_sql,
378 tenant_column: spec.tenant_column,
379 }
380}
381
382/// Every identifier a view splices into SQL, quoted.
383///
384/// Field names match [`SqlView`]'s so the statement builders below can
385/// destructure this in place of the view and leave their SQL untouched.
386struct Quoted {
387 child_table: String,
388 child_pk: String,
389 fk_column: String,
390 parent_table: String,
391 parent_pk: String,
392 counter_column: String,
393}
394
395fn quoted(view: &SqlView) -> Quoted {
396 Quoted {
397 child_table: quote_ident(view.child_table),
398 child_pk: quote_ident(view.child_pk),
399 fk_column: quote_ident(view.fk_column),
400 parent_table: quote_ident(view.parent_table),
401 parent_pk: quote_ident(view.parent_pk),
402 counter_column: quote_ident(view.counter_column),
403 }
404}
405
406/// What the `{bin}` placeholder in a lowered string comparison resolves to.
407///
408/// A derivation filter compares a string field with `==` or `!=`, and the Rust
409/// lowering compares bytes. SQL compares by the column's collation, which on a
410/// `NOCASE` or otherwise case-folding column would call `"PUB"` and `'pub'`
411/// equal where Rust does not, so the record paths and the set-based paths of
412/// one derivation would disagree. Pinning the comparison to the backend's
413/// bytewise collation makes both sides mean the same thing whatever the column
414/// was declared with.
415#[cfg(not(feature = "sqlite"))]
416const BINARY_COLLATION: &str = "COLLATE \"C\"";
417#[cfg(feature = "sqlite")]
418const BINARY_COLLATION: &str = "COLLATE BINARY";
419
420/// Resolve the `{c}` child-alias and `{bin}` collation placeholders in a
421/// lowered SQL fragment.
422///
423/// One lowered filter has to serve statements that alias the child table
424/// differently (`__autumn_cc_child`, `__autumn_cc_child_t`), so `#[model]`
425/// emits the alias as a placeholder and each statement substitutes its own.
426/// The collation is a placeholder for the same reason in the other direction:
427/// the fragment is lowered once, and which backend runs it is a feature flag.
428fn with_alias(sql: &str, alias: &str) -> String {
429 let out = sql.replace("{c}", alias).replace("{bin}", BINARY_COLLATION);
430 assert!(
431 !out.contains('{'),
432 "a lowered fragment carries an unresolved placeholder: {out}"
433 );
434 out
435}
436
437/// The derivation's row filter for `alias`, or `""` for a counter cache.
438fn filter_predicate(view: &SqlView, alias: &str) -> String {
439 with_alias(view.filter_sql, alias)
440}
441
442/// One row's contribution expression for `alias`.
443fn contrib_expr(view: &SqlView, alias: &str) -> String {
444 with_alias(view.contrib_sql, alias)
445}
446
447/// The aggregate that folds a set of child rows into the maintained value.
448///
449/// `COUNT(*)` for a count: the filter is in the surrounding `WHERE`, so every
450/// counted row already qualifies. `COALESCE(SUM(...), 0)` for a weighted sum,
451/// because `SUM` over an empty set is NULL and the maintained column is not.
452fn aggregate_expr(view: &SqlView, alias: &str) -> String {
453 if view.counts_rows() {
454 "COUNT(*)".to_owned()
455 } else {
456 format!("COALESCE(SUM({}), 0)", contrib_expr(view, alias))
457 }
458}
459
460/// `CASE WHEN <filter> THEN <contrib> ELSE 0 END`, for the statements that read
461/// a row's contribution as a column.
462///
463/// The `1 = 1` seed absorbs `filter_sql`'s leading ` AND `, which every other
464/// caller concatenates onto a predicate it already has.
465///
466/// The `CAST` is not cosmetic: this value is *decoded* into an `i64`, and a bare
467/// integer literal is `integer` on Postgres, so the driver would refuse the
468/// four bytes it got where it expected eight. Casting also normalises a legacy
469/// 32-bit summed column to the width the maintained value is read at.
470fn contrib_case_expr(view: &SqlView, alias: &str) -> String {
471 let contrib = contrib_expr(view, alias);
472 if view.filter_sql.is_empty() {
473 return format!("CAST({contrib} AS BIGINT)");
474 }
475 let filter = filter_predicate(view, alias);
476 format!("CAST(CASE WHEN 1 = 1{filter} THEN {contrib} ELSE 0 END AS BIGINT)")
477}
478
479/// `AND <parent>.<tenant> = __autumn_cc_child.<tenant>`, for the statements that
480/// already have the child row joined in a sub-select. Empty when the association
481/// declares no `counter_cache_tenant`.
482///
483/// The predicate must be applied to **every** delta path or it makes things
484/// worse rather than better: scoping only the increment would leave the matching
485/// decrement unscoped, so a cross-tenant foreign key would drive another
486/// tenant's counter down without ever driving it up.
487///
488/// `AND EXISTS (SELECT 1 FROM <child> x WHERE x.<pk> = N AND <parent>.tenant IS
489/// NOT DISTINCT FROM x.tenant)` for the parent-keyed statements, which have no
490/// child row to join, empty otherwise. The `EXISTS` (rather than a scalar
491/// sub-select) keeps the statement a no-op when the child row is absent, which
492/// is what a missing child has always meant here.
493///
494/// Deliberately a sub-select on the child row rather than a bound ambient
495/// tenant: the invariant being enforced is "the parent is in the same tenant as
496/// the child that names it", which is exactly what makes a cross-tenant foreign
497/// key move nothing. It needs no tenant plumbing through the generated code, and
498/// it is correct for `across_tenants()` callers too.
499fn tenant_predicate_joined(view: &SqlView) -> String {
500 let Some(tenant_column) = view.tenant_column else {
501 return String::new();
502 };
503 let tenant_column = quote_ident(tenant_column);
504 let parent_table = quote_ident(view.parent_table);
505 format!(
506 " AND {parent_table}.{tenant_column} {IS_NOT_DISTINCT_FROM} \
507 {CHILD_ALIAS}.{tenant_column}"
508 )
509}
510
511/// `AND CAST(<parent>.<tenant> AS TEXT) IS NOT DISTINCT FROM $3`, for a delta
512/// scoped by a tenant the pre-update capture read (see
513/// [`TenantScope::Captured`]). Both sides are text so one bind serves every
514/// tenant column type; `IS NOT DISTINCT FROM` matches a NULL tenant to a NULL
515/// bind the way the joined predicate does.
516fn tenant_predicate_captured(view: &SqlView) -> String {
517 let Some(tenant_column) = view.tenant_column else {
518 return String::new();
519 };
520 let tenant_column = quote_ident(tenant_column);
521 let parent_table = quote_ident(view.parent_table);
522 format!(" AND CAST({parent_table}.{tenant_column} AS TEXT) {IS_NOT_DISTINCT_FROM} {PH3}")
523}
524
525fn tenant_predicate(view: &SqlView, child_id: i64) -> String {
526 let Some(tenant_column) = view.tenant_column else {
527 return String::new();
528 };
529 let tenant_column = quote_ident(tenant_column);
530 let Quoted {
531 child_table,
532 child_pk,
533 parent_table,
534 ..
535 } = quoted(view);
536 format!(
537 " AND EXISTS \
538 (SELECT 1 FROM {child_table} AS {CHILD_ALIAS}_t \
539 WHERE {CHILD_ALIAS}_t.{child_pk} = {child_id} \
540 AND {parent_table}.{tenant_column} {IS_NOT_DISTINCT_FROM} \
541 {CHILD_ALIAS}_t.{tenant_column})"
542 )
543}
544
545/// The left-hand side of a `SET c = <acc> +/- ...` adjustment.
546///
547/// A plain counter cache reads the column bare, so its SQL stays byte-identical
548/// to what it was before derivations existed. A derivation wraps it in
549/// `COALESCE(c, 0)`: a derivation is adopted on an existing table, where the
550/// column can still be NULL, and `NULL +/- expr` is NULL. One such adjustment
551/// would poison the column, and a NULL maintained value is the one outcome no
552/// repair can tell apart from legitimate drift.
553fn accumulator(view: &SqlView, counter_column: &str) -> String {
554 if view.is_plain() {
555 counter_column.to_owned()
556 } else {
557 format!("COALESCE({counter_column}, 0)")
558 }
559}
560
561/// `AND __autumn_cc_child.deleted_at IS NULL` (or `IS NOT NULL`) for a
562/// soft-deleting child, empty for a child with no `deleted_at` column.
563fn live_predicate(view: &SqlView, want_live: bool) -> String {
564 if !view.child_soft_delete {
565 return String::new();
566 }
567 let op = if want_live { "IS NULL" } else { "IS NOT NULL" };
568 let deleted_at = quote_ident(DELETED_AT);
569 format!(" AND {CHILD_ALIAS}.{deleted_at} {op}")
570}
571
572/// Take, for the rest of the transaction, the lock that serializes every
573/// counter-cached mutation on a table with a leg onto itself.
574///
575/// Call it before a raw write of your own on such a table, ahead of the
576/// `INSERT`, `UPDATE` or `DELETE` itself, and before
577/// [`counter_cache_after_insert_by_id`] or its siblings; see the note on that
578/// function.
579///
580/// Onto its own table, a mutation locks a child row and then a parent row of
581/// the same table. Two such mutations can want each other's rows: one moves
582/// node A under B (holds A, waits for B) while another moves B under A (holds
583/// B, waits for A), and the database breaks the cycle by aborting one, which
584/// the generated repository transaction does not retry. Ordering the parent
585/// deltas cannot help, since each child lock is already held. So every such
586/// mutation first takes one transaction-scoped advisory lock keyed by the
587/// table, before any row lock, and two of them take turns instead of crossing.
588/// Every hook here takes it on entry, and every generated mutation that
589/// row-locks before its hook runs (`upsert_many`'s `FOR UPDATE` load, the
590/// hooked delete's load for `before_delete`, `delete_many`'s preload, the
591/// dependent cascade's child lock, a retention sweep's batch lock) calls this
592/// at the top of its transaction too, so the advisory lock always comes before
593/// the first row lock; the hook's later take is then a re-entrant no-op. A
594/// table without a self-referential leg costs nothing here. `SQLite`
595/// serializes writers with `BEGIN IMMEDIATE` already, so it has nothing to
596/// take.
597///
598/// # Errors
599///
600/// Propagates any database error from the lock statement.
601pub async fn counter_cache_serialize_self_referential<M: 'static>(
602 conn: &mut RuntimeConnection,
603 specs: &[CounterCacheSpec<M>],
604) -> AutumnResult<()> {
605 for table in self_referential_tables(specs) {
606 serialize_table(conn, table).await?;
607 }
608 Ok(())
609}
610
611/// The tables of `specs` with a leg onto themselves, each once, in a stable
612/// order (so two mutations over several such tables take the locks in the
613/// same order).
614fn self_referential_tables<M: 'static>(specs: &[CounterCacheSpec<M>]) -> Vec<&'static str> {
615 let mut tables: Vec<&'static str> = specs
616 .iter()
617 .filter(|spec| spec.parent_table == spec.child_table)
618 .map(|spec| spec.child_table)
619 .collect();
620 tables.sort_unstable();
621 tables.dedup();
622 tables
623}
624
625#[cfg(not(feature = "sqlite"))]
626async fn serialize_table(conn: &mut RuntimeConnection, table: &str) -> AutumnResult<()> {
627 // The two-key form keeps this out of the single-key namespace the
628 // framework's migration and job locks use; the first key names this
629 // purpose, the second the table.
630 diesel::sql_query(
631 "SELECT pg_advisory_xact_lock(hashtext('autumn_counter_cache'), hashtext($1))",
632 )
633 .bind::<Text, _>(table)
634 .execute(conn)
635 .await
636 .map_err(AutumnError::from)?;
637 Ok(())
638}
639
640#[cfg(feature = "sqlite")]
641#[allow(clippy::unused_async, reason = "one call shape for both backends")]
642async fn serialize_table(_conn: &mut RuntimeConnection, _table: &str) -> AutumnResult<()> {
643 Ok(())
644}
645
646/// Apply `delta` to one parent's counter with a single atomic statement.
647///
648/// This is the primitive AC5 rests on: `SET c = c + $1` is resolved by the
649/// database, so N concurrent callers commute and none can lose another's update
650/// the way a `SELECT` + `UPDATE` round trip would.
651///
652/// # Errors
653///
654/// Propagates any database error from the `UPDATE`.
655pub async fn counter_cache_apply_delta<M: 'static>(
656 conn: &mut RuntimeConnection,
657 spec: &CounterCacheSpec<M>,
658 parent_id: i64,
659 delta: i64,
660 scope: TenantScope,
661) -> AutumnResult<()> {
662 let view = view(spec);
663 assert_spec_idents(&view);
664 let Quoted {
665 parent_table,
666 parent_pk,
667 counter_column,
668 ..
669 } = quoted(&view);
670 let tenant = match &scope {
671 TenantScope::SameTenantAsChild(child_id) => tenant_predicate(&view, *child_id),
672 TenantScope::Captured(_) => tenant_predicate_captured(&view),
673 TenantScope::Unscoped => String::new(),
674 };
675 let accumulator = accumulator(&view, &counter_column);
676 let sql = format!(
677 "UPDATE {parent_table} SET {counter_column} = {accumulator} + {PH1} \
678 WHERE {parent_table}.{parent_pk} = {PH2}{tenant}"
679 );
680 let query = diesel::sql_query(sql)
681 .bind::<BigInt, _>(delta)
682 .bind::<BigInt, _>(parent_id);
683 // The captured tenant is a third bind, but only when the leg has a tenant
684 // column: without one the predicate is empty and so is the bind list.
685 if let (TenantScope::Captured(tenant), true) = (scope, view.tenant_column.is_some()) {
686 query
687 .bind::<Nullable<Text>, _>(tenant)
688 .execute(conn)
689 .await
690 .map_err(AutumnError::from)?;
691 return Ok(());
692 }
693 query.execute(conn).await.map_err(AutumnError::from)?;
694 Ok(())
695}
696
697/// Apply `delta` to the parent of the child row `child_id`, resolving the parent
698/// through a sub-select so no round trip is needed and a `NULL` foreign key is a
699/// no-op (`IN (NULL)` matches nothing).
700///
701/// `require_live` scopes the sub-select to a live child (`deleted_at IS NULL`)
702/// for a soft-deleting model, which is what makes a repeated soft delete — or a
703/// hard delete of an already-soft-deleted row — decrement exactly once.
704/// `require_soft_deleted` is its mirror, used by `restore`.
705///
706/// For a `#[derivation]` the magnitude cannot come from the caller: the weight
707/// is a property of the child row, which only the database can see here. So
708/// `delta` supplies the **sign** and the statement reads the contribution
709/// itself. The filter sits in both sub-selects, so a row the filter rejects
710/// matches no parent and the statement is a no-op.
711///
712/// # Errors
713///
714/// Propagates any database error from the `UPDATE`.
715#[doc(hidden)]
716pub async fn counter_cache_apply_delta_by_child_id<M: 'static>(
717 conn: &mut RuntimeConnection,
718 spec: &CounterCacheSpec<M>,
719 child_id: i64,
720 delta: i64,
721 child_state: ChildState,
722) -> AutumnResult<()> {
723 let view = view(spec);
724 assert_spec_idents(&view);
725 let state_predicate = match child_state {
726 ChildState::Any => String::new(),
727 ChildState::Live => live_predicate(&view, true),
728 ChildState::SoftDeleted => live_predicate(&view, false),
729 };
730 // The parent is resolved by a sub-select on the child row, so the tenant
731 // check is a correlated comparison in the outer `WHERE`. It names the child
732 // alias, which is only in scope for a sub-select the outer statement
733 // correlates with, so the whole predicate moves into a second sub-select.
734 let tenant = tenant_predicate(&view, child_id);
735
736 if !view.is_plain() {
737 debug_assert!(
738 delta == 1 || delta == -1,
739 "a derivation delta carries only a sign; the weight comes from the row"
740 );
741 let sql =
742 weighted_delta_by_child_id_sql(&view, child_id, delta < 0, &state_predicate, &tenant);
743 diesel::sql_query(sql)
744 .execute(conn)
745 .await
746 .map_err(AutumnError::from)?;
747 return Ok(());
748 }
749
750 let Quoted {
751 child_table,
752 child_pk,
753 fk_column,
754 parent_table,
755 parent_pk,
756 counter_column,
757 ..
758 } = quoted(&view);
759 let sql = format!(
760 "UPDATE {parent_table} SET {counter_column} = {counter_column} + {PH1} \
761 WHERE {parent_table}.{parent_pk} IN \
762 (SELECT {CHILD_ALIAS}.{fk_column} FROM {child_table} AS {CHILD_ALIAS} \
763 WHERE {CHILD_ALIAS}.{child_pk} = {PH2} \
764 AND {CHILD_ALIAS}.{fk_column} IS NOT NULL{state_predicate}{FOR_UPDATE}){tenant}"
765 );
766 diesel::sql_query(sql)
767 .bind::<BigInt, _>(delta)
768 .bind::<BigInt, _>(child_id)
769 .execute(conn)
770 .await
771 .map_err(AutumnError::from)?;
772 Ok(())
773}
774
775/// The `#[derivation]` form of [`counter_cache_apply_delta_by_child_id`]: the
776/// weight is read from the child row instead of bound by the caller.
777///
778/// `child_id` is inlined rather than bound, so the two occurrences stay one
779/// statement on both backends (`SQLite`'s `?` is positional, so a second bind
780/// would be required there and not on Postgres). It is an `i64`, so its decimal
781/// rendering cannot contain SQL syntax, the same type-level guarantee
782/// [`id_list`] rests on.
783///
784/// `COALESCE` is belt-and-braces: the outer `WHERE` already restricts the
785/// statement to parents whose child row qualifies, so the sub-select cannot be
786/// empty. But a NULL in the summed column would still poison the column, and a
787/// NULL maintained value is the one outcome no repair can distinguish from
788/// legitimate drift.
789fn weighted_delta_by_child_id_sql(
790 view: &SqlView,
791 child_id: i64,
792 subtract: bool,
793 state_predicate: &str,
794 tenant: &str,
795) -> String {
796 let Quoted {
797 child_table,
798 child_pk,
799 fk_column,
800 parent_table,
801 parent_pk,
802 counter_column,
803 ..
804 } = quoted(view);
805 let filter = filter_predicate(view, CHILD_ALIAS);
806 let contrib = contrib_expr(view, CHILD_ALIAS);
807 let sign = if subtract { "-" } else { "+" };
808 let accumulator = accumulator(view, &counter_column);
809 format!(
810 "UPDATE {parent_table} SET {counter_column} = {accumulator} {sign} \
811 COALESCE((SELECT {contrib} FROM {child_table} AS {CHILD_ALIAS} \
812 WHERE {CHILD_ALIAS}.{child_pk} = {child_id} \
813 AND {CHILD_ALIAS}.{fk_column} IS NOT NULL{state_predicate}{filter}), 0) \
814 WHERE {parent_table}.{parent_pk} IN \
815 (SELECT {CHILD_ALIAS}.{fk_column} FROM {child_table} AS {CHILD_ALIAS} \
816 WHERE {CHILD_ALIAS}.{child_pk} = {child_id} \
817 AND {CHILD_ALIAS}.{fk_column} \
818 IS NOT NULL{state_predicate}{filter}{FOR_UPDATE}){tenant}"
819 )
820}
821
822/// How a parent-keyed delta is confined to a tenant.
823#[derive(Debug, Clone, PartialEq, Eq)]
824pub enum TenantScope {
825 /// Require the parent to sit in the same tenant as this child row. Emits no
826 /// predicate for a child model without a tenant column.
827 SameTenantAsChild(i64),
828 /// Require the parent to sit in this tenant, spelled as the text the
829 /// pre-update capture read (`None` for a NULL tenant). Used for the old
830 /// side of a child that changed tenant: its row now says the new tenant,
831 /// so the live predicate would miss the parent it is leaving. Emits no
832 /// predicate for a child model without a tenant column.
833 Captured(Option<String>),
834 /// No tenant predicate. Used only by paths that have no child row to scope
835 /// against (`recompute`, which sweeps the parent table wholesale).
836 Unscoped,
837}
838
839/// Which soft-delete state the child row must be in for a by-id delta to apply.
840#[doc(hidden)]
841#[derive(Debug, Clone, Copy, PartialEq, Eq)]
842pub enum ChildState {
843 /// No `deleted_at` predicate at all.
844 Any,
845 /// Only when the child is live (`deleted_at IS NULL`). Used by the
846 /// decrement, so a repeated soft delete moves nothing.
847 Live,
848 /// Only when the child is soft-deleted. Used by `restore`, so restoring an
849 /// already-live row moves nothing.
850 SoftDeleted,
851}
852
853/// Increment every counter-cached parent of a freshly inserted child record.
854///
855/// A record that arrives already soft-deleted moves nothing: the counter is
856/// defined as live rows only, and every other path agrees on that, so counting a
857/// born-deleted row would inflate the counter until the next repair. An ordinary
858/// insert is live, so this changes nothing for it.
859///
860/// # Errors
861///
862/// Propagates any database error from the `UPDATE`s.
863#[doc(hidden)]
864pub async fn counter_cache_after_insert<M: Send + Sync + 'static>(
865 conn: &mut RuntimeConnection,
866 specs: &[CounterCacheSpec<M>],
867 record: &M,
868) -> AutumnResult<()> {
869 counter_cache_serialize_self_referential(conn, specs).await?;
870 let mut contributions: Vec<Contribution> = Vec::with_capacity(specs.len());
871 for (index, spec) in specs.iter().enumerate() {
872 if !(spec.live_of)(record) {
873 continue;
874 }
875 // A row the derivation's filter rejects contributes 0, and a 0 delta is
876 // no statement at all, not even a `+ 0` write to the parent row.
877 let contrib = (spec.contrib_of)(record);
878 if contrib == 0 {
879 continue;
880 }
881 if let Some(parent_id) = (spec.fk_of)(record) {
882 let scope = TenantScope::SameTenantAsChild((spec.pk_of)(record));
883 contributions.push((index, parent_id, contrib, scope));
884 }
885 }
886 apply_ordered(conn, specs, contributions).await
887}
888
889/// [`counter_cache_after_insert`] over a batch of records (`save_many`).
890///
891/// Children sharing a parent are folded into **one** `+ n` statement per parent,
892/// so inserting a 1000-row chunk issues one `UPDATE` per distinct parent rather
893/// than one per row — still a single atomic statement each, so the concurrency
894/// guarantee is unchanged.
895///
896/// # Errors
897///
898/// Propagates any database error from the `UPDATE`s.
899#[doc(hidden)]
900pub async fn counter_cache_after_insert_many<M: Send + Sync + 'static>(
901 conn: &mut RuntimeConnection,
902 specs: &[CounterCacheSpec<M>],
903 records: &[M],
904) -> AutumnResult<()> {
905 counter_cache_serialize_self_referential(conn, specs).await?;
906 let mut contributions: Vec<Contribution> = Vec::new();
907 for (index, spec) in specs.iter().enumerate() {
908 for record in records {
909 if !(spec.live_of)(record) {
910 continue;
911 }
912 let contrib = (spec.contrib_of)(record);
913 if contrib == 0 {
914 continue;
915 }
916 if let Some(parent_id) = (spec.fk_of)(record) {
917 let scope = TenantScope::SameTenantAsChild((spec.pk_of)(record));
918 contributions.push((index, parent_id, contrib, scope));
919 }
920 }
921 }
922 apply_ordered(conn, specs, contributions).await
923}
924
925/// One parent row a mutation will move: `(spec index, parent id, delta, how
926/// the parent is confined to a tenant)`.
927type Contribution = (usize, i64, i64, TenantScope);
928
929/// One leg's pre-mutation view of a child row.
930#[derive(Debug, Clone, PartialEq, Eq)]
931pub struct Captured {
932 /// The parent the row contributes to.
933 pub parent: i64,
934 /// Its weight in the aggregate. `0` for a row the filter rejects: it *has* a
935 /// parent, and it weighs nothing, which is what makes a filter flip on an
936 /// unchanged parent visible as a delta.
937 pub contrib: i64,
938 /// The row's tenant as text, for a leg that reads it (see
939 /// [`CounterCacheSpec::tenant_of`]); `None` otherwise, or for a NULL tenant.
940 pub tenant: Option<String>,
941}
942
943/// One leg's pre-mutation [`Captured`] for a child row.
944///
945/// `None` when the row contributes to no parent at all: it does not exist, its
946/// foreign key is NULL, or it is soft-deleted.
947pub type CapturedContribution = Option<Captured>;
948
949/// Every leg's [`CapturedContribution`] for a batch of children, keyed by child
950/// primary key and sorted by it, as the bulk update paths consume it.
951pub type CapturedContributions = Vec<(i64, Vec<CapturedContribution>)>;
952
953/// Apply every contribution in the lock order all counter-cached mutations
954/// agree on: `(parent_table, parent_id)`.
955///
956/// Ordering by parent id alone is not enough once a mutation moves more than one
957/// leg. Two child models that declare their legs in opposite orders — one
958/// `belongs_to(User)` then `belongs_to(Post)`, the other the reverse — would
959/// take the same two row locks in opposite orders and deadlock, and generated
960/// repository transactions are single-attempt, so one request fails outright.
961/// Keying on the parent *table* as well makes the order a property of the schema
962/// rather than of any one model's declaration order, so every writer in the
963/// process agrees on it — including two legs that point at the same table, which
964/// a per-spec ordering cannot reconcile.
965///
966/// Folding several children of one parent into a single `+ n` is a pure
967/// optimization: re-parenting 1000 children becomes two statements rather than
968/// 2000. It is only sound when the association has no tenant column, since the
969/// tenant predicate is scoped to a single witness child and a mixed-tenant batch
970/// folded behind one arbitrary witness would either sweep cross-tenant children
971/// into the increment or drop legitimate ones. Tenant-scoped specs therefore
972/// keep one statement per contribution — in the same global order.
973async fn apply_ordered<M: 'static>(
974 conn: &mut RuntimeConnection,
975 specs: &[CounterCacheSpec<M>],
976 contributions: Vec<Contribution>,
977) -> AutumnResult<()> {
978 if contributions.is_empty() {
979 return Ok(());
980 }
981 for (spec_index, parent_id, delta, scope) in fold_and_order(specs, contributions) {
982 counter_cache_apply_delta(conn, &specs[spec_index], parent_id, delta, scope).await?;
983 }
984 Ok(())
985}
986
987/// The pure half of [`apply_ordered`]: fold, drop the no-ops, and put what is
988/// left in the global lock order. Split out so the ordering is unit-testable
989/// without a database.
990fn fold_and_order<M: 'static>(
991 specs: &[CounterCacheSpec<M>],
992 contributions: Vec<Contribution>,
993) -> Vec<Contribution> {
994 // The running total is an `i128`, so weights that cancel do cancel even
995 // when no `i64` could hold an intermediate sum (`[MAX, MAX, -MAX, -MAX]`
996 // is zero, not two statements the database would refuse).
997 let mut totals: Vec<(usize, i64, i128, TenantScope)> = Vec::with_capacity(contributions.len());
998 // `(spec, parent)` -> where that parent's running total lives in `totals`.
999 let mut seen: HashMap<(usize, i64), usize> = HashMap::new();
1000 for (spec_index, parent_id, delta, scope) in contributions {
1001 if specs[spec_index].tenant_column.is_some() {
1002 totals.push((spec_index, parent_id, i128::from(delta), scope));
1003 continue;
1004 }
1005 if let Some(&at) = seen.get(&(spec_index, parent_id)) {
1006 totals[at].2 += i128::from(delta);
1007 } else {
1008 seen.insert((spec_index, parent_id), totals.len());
1009 totals.push((spec_index, parent_id, i128::from(delta), scope));
1010 }
1011 }
1012 // A net total that no single `i64` delta can carry goes out as several
1013 // deltas of the same sign, so the parent moves monotonically from where it
1014 // is to where it ends: when both are representable, so is every step.
1015 // Zero totals are dropped rather than issued as `+ 0`.
1016 let mut folded: Vec<Contribution> = Vec::with_capacity(totals.len());
1017 for (spec_index, parent_id, mut total, scope) in totals {
1018 while total != 0 {
1019 let chunk = total.clamp(i128::from(i64::MIN), i128::from(i64::MAX));
1020 #[allow(clippy::cast_possible_truncation)]
1021 folded.push((spec_index, parent_id, chunk as i64, scope.clone()));
1022 total -= chunk;
1023 }
1024 }
1025 folded
1026 .sort_by_key(|&(spec_index, parent_id, _, _)| (specs[spec_index].parent_table, parent_id));
1027 order_tenant_runs(specs, &mut folded);
1028 folded
1029}
1030
1031/// Order each tenant-scoped parent's deltas so their running sum stays between
1032/// zero and their total.
1033///
1034/// A tenant-scoped leg cannot fold: each delta is guarded by its own child's
1035/// tenant, and two children of one parent need not share one. What it can do
1036/// is choose the order. Walking toward the total and stepping back past it
1037/// keeps every prefix within `[min(0, total), max(0, total)]` whenever the
1038/// deltas allow it at all, so `[MAX, MAX, -MAX]` goes out as `MAX, -MAX, MAX`
1039/// and a parent whose start and end both fit never passes through a value
1040/// that does not. The sort is stable, so the global lock order is unchanged.
1041fn order_tenant_runs<M: 'static>(specs: &[CounterCacheSpec<M>], folded: &mut [Contribution]) {
1042 let mut start = 0;
1043 while start < folded.len() {
1044 let (spec_index, parent_id) = (folded[start].0, folded[start].1);
1045 let mut end = start + 1;
1046 while end < folded.len() && folded[end].0 == spec_index && folded[end].1 == parent_id {
1047 end += 1;
1048 }
1049 if specs[spec_index].tenant_column.is_some() && end - start > 1 {
1050 let run = &mut folded[start..end];
1051 let total: i128 = run.iter().map(|(_, _, delta, _)| i128::from(*delta)).sum();
1052 let mut remaining: Vec<Contribution> = run.to_vec();
1053 let mut prefix: i128 = 0;
1054 for slot in run.iter_mut() {
1055 let toward_total = prefix < total || (prefix == total && total < 0);
1056 // Ties keep their original (child) order, so equal weights go
1057 // out as they were captured and the mutation stays
1058 // deterministic: `min_by_key` takes the first minimum, and the
1059 // reversed scan makes `max_by_key` take the first maximum.
1060 let pick = if toward_total {
1061 (0..remaining.len()).rev().max_by_key(|&i| remaining[i].2)
1062 } else {
1063 (0..remaining.len()).min_by_key(|&i| remaining[i].2)
1064 }
1065 .expect("a run has at least one delta left per slot");
1066 let next = remaining.remove(pick);
1067 prefix += i128::from(next.2);
1068 *slot = next;
1069 }
1070 }
1071 start = end;
1072 }
1073}
1074
1075/// Spec indices in the same global order, for the paths that resolve the parent
1076/// inside the statement and so cannot key on its id.
1077///
1078/// This removes every cross-table cycle, which is the half these paths can
1079/// reach: a by-id call touches one parent per leg, so two legs onto *different*
1080/// tables are ordered here, and two legs onto the *same* table are the residual
1081/// (`recompute` is the repair, and a deadlock there surfaces as an error rather
1082/// than as drift). `counter_column` breaks ties so the order is total.
1083fn specs_in_lock_order<M: 'static>(specs: &[CounterCacheSpec<M>]) -> Vec<usize> {
1084 let mut order: Vec<usize> = (0..specs.len()).collect();
1085 order.sort_by_key(|&i| (specs[i].parent_table, specs[i].counter_column));
1086 order
1087}
1088
1089/// Increment every counter-cached parent of the child row `child_id`.
1090///
1091/// The documented escape hatch for applications that insert a child with their
1092/// own SQL inside their own transaction (rather than through the generated
1093/// repository) and still want the framework to own the counter arithmetic:
1094///
1095/// ```rust,ignore
1096/// counter_cache_serialize_self_referential(conn, Comment::counter_caches()).await?;
1097/// let id = diesel::insert_into(comments::table) /* … */ .get_result(conn).await?;
1098/// counter_cache_after_insert_by_id(conn, Comment::counter_caches(), id).await?;
1099/// ```
1100///
1101/// The first line matters on a table with a leg onto itself, and costs nothing
1102/// on any other: every generated mutation takes that advisory lock before its
1103/// first row lock, and an inserted row is locked from the moment it exists (a
1104/// concurrent upsert of the same id waits on it through the primary key), so
1105/// a raw insert that takes the lock only here, after the row, can hold the
1106/// row while a generated mutation holds the lock and waits for the row, and
1107/// the database aborts one of them. Take the lock, then write.
1108///
1109/// For a soft-deleting child the increment is conditional on the row being live,
1110/// mirroring the delete side. Raw SQL can insert a row with `deleted_at` already
1111/// set, and every other path — delete, update, recompute — defines the counter as
1112/// live rows only, so incrementing for a born-deleted row would inflate the
1113/// counter until the next repair.
1114///
1115/// # Errors
1116///
1117/// Propagates any database error from the `UPDATE`s.
1118pub async fn counter_cache_after_insert_by_id<M: 'static>(
1119 conn: &mut RuntimeConnection,
1120 specs: &[CounterCacheSpec<M>],
1121 child_id: i64,
1122) -> AutumnResult<()> {
1123 counter_cache_serialize_self_referential(conn, specs).await?;
1124 for index in specs_in_lock_order(specs) {
1125 let spec = &specs[index];
1126 let state = if spec.child_soft_delete {
1127 ChildState::Live
1128 } else {
1129 ChildState::Any
1130 };
1131 counter_cache_apply_delta_by_child_id(conn, spec, child_id, 1, state).await?;
1132 }
1133 Ok(())
1134}
1135
1136/// Decrement every counter-cached parent of the child row `child_id`.
1137///
1138/// Must be called **before** the row is deleted or soft-deleted: it resolves the
1139/// parent from the still-present child row, and (for a soft-deleting model)
1140/// requires that row to still be live, so the decrement happens exactly once
1141/// however many times the delete is retried.
1142///
1143/// # Errors
1144///
1145/// Propagates any database error from the `UPDATE`s.
1146pub async fn counter_cache_before_delete_by_id<M: 'static>(
1147 conn: &mut RuntimeConnection,
1148 specs: &[CounterCacheSpec<M>],
1149 child_id: i64,
1150) -> AutumnResult<()> {
1151 counter_cache_serialize_self_referential(conn, specs).await?;
1152 for index in specs_in_lock_order(specs) {
1153 let spec = &specs[index];
1154 let state = if spec.child_soft_delete {
1155 ChildState::Live
1156 } else {
1157 ChildState::Any
1158 };
1159 counter_cache_apply_delta_by_child_id(conn, spec, child_id, -1, state).await?;
1160 }
1161 Ok(())
1162}
1163
1164/// Take the child row locks a bulk decrement needs, in ascending id order.
1165///
1166/// The returned rows are discarded — the statement exists for its locks (see
1167/// [`counter_cache_before_delete_many`]'s lock-order note).
1168fn child_lock_sql(view: &SqlView, id_list: &str) -> String {
1169 let Quoted {
1170 child_table,
1171 child_pk,
1172 ..
1173 } = quoted(view);
1174 format!(
1175 "SELECT {child_pk} AS id FROM {child_table} \
1176 WHERE {child_pk} IN ({id_list}) ORDER BY {child_pk}{FOR_UPDATE}"
1177 )
1178}
1179
1180/// [`counter_cache_before_delete_by_id`] over a batch of ids (`delete_many`).
1181///
1182/// Issues **one** statement per spec regardless of batch size: each affected
1183/// parent's counter drops by the number of its children in the batch, computed
1184/// by the database. As with the single-row form, this must run *before* the rows
1185/// are deleted, and for a soft-deleting model it counts only rows that are still
1186/// live so a repeated bulk delete moves nothing.
1187///
1188/// # Lock order
1189///
1190/// The children are locked in ascending id order **before** any parent counter
1191/// is touched. Every other path in this module reaches the parent through a
1192/// child row it has already locked (`counter_cache_apply_delta_by_child_id`
1193/// carries `FOR UPDATE` on its sub-select; the update paths lock the child while
1194/// capturing its foreign keys), so the module's lock order is uniformly
1195/// child-then-parent. Without this statement the bulk path would invert it —
1196/// locking the parent for the decrement here and the child only later, when the
1197/// `DELETE` itself runs — and a bulk delete racing an `update` that re-parents
1198/// one of the same children would deadlock, which generated repository
1199/// transactions do not retry.
1200///
1201/// # Errors
1202///
1203/// Propagates any database error from the `SELECT` or the `UPDATE`s.
1204#[doc(hidden)]
1205pub async fn counter_cache_before_delete_many<M: 'static>(
1206 conn: &mut RuntimeConnection,
1207 specs: &[CounterCacheSpec<M>],
1208 child_ids: &[i64],
1209) -> AutumnResult<()> {
1210 counter_cache_serialize_self_referential(conn, specs).await?;
1211 if specs.is_empty() || child_ids.is_empty() {
1212 return Ok(());
1213 }
1214 let id_list = id_list(child_ids);
1215
1216 // One statement, not one per spec: every spec here is a leg of the *same*
1217 // child model, so they all name the same table and primary key.
1218 let first = view(&specs[0]);
1219 assert_spec_idents(&first);
1220 diesel::sql_query(child_lock_sql(&first, &id_list))
1221 .load::<IdRow>(conn)
1222 .await
1223 .map_err(AutumnError::from)?;
1224
1225 for index in specs_in_lock_order(specs) {
1226 let view = view(&specs[index]);
1227 assert_spec_idents(&view);
1228 let sql = bulk_decrement_sql(&view, &id_list);
1229 diesel::sql_query(sql)
1230 .execute(conn)
1231 .await
1232 .map_err(AutumnError::from)?;
1233 }
1234 Ok(())
1235}
1236
1237/// [`counter_cache_before_delete_many`] restricted to the ONE leg whose foreign
1238/// key is about to be cleared (`dependent = nullify`).
1239///
1240/// A nullify detaches the children from a single parent association; their other
1241/// counter-cached legs are untouched, so decrementing every spec would
1242/// permanently undercount them. For example, nullifying `comments.author_id`
1243/// when a user is deleted must not drop `posts.comment_count` for comments that
1244/// remain attached to their post.
1245///
1246/// # Errors
1247///
1248/// Propagates any database error from the `UPDATE`s.
1249#[doc(hidden)]
1250pub async fn counter_cache_before_detach_many<M: 'static>(
1251 conn: &mut RuntimeConnection,
1252 specs: &[CounterCacheSpec<M>],
1253 fk_column: &str,
1254 child_ids: &[i64],
1255) -> AutumnResult<()> {
1256 counter_cache_serialize_self_referential(conn, specs).await?;
1257 let detached: Vec<CounterCacheSpec<M>> = specs
1258 .iter()
1259 .filter(|spec| spec.fk_column == fk_column)
1260 .copied()
1261 .collect();
1262 counter_cache_before_delete_many(conn, &detached, child_ids).await
1263}
1264
1265/// The one `UPDATE` a bulk decrement issues per leg.
1266///
1267/// The maintained value drops by the aggregate of the batch's qualifying child
1268/// rows, computed by the database in one statement. Both sub-selects carry the
1269/// derivation filter, so a batch of rows the filter rejects matches no parent
1270/// and writes nothing, which is also what keeps NULL arithmetic out of reach.
1271fn bulk_decrement_sql(view: &SqlView, id_list: &str) -> String {
1272 let Quoted {
1273 child_table,
1274 child_pk,
1275 fk_column,
1276 parent_table,
1277 parent_pk,
1278 counter_column,
1279 ..
1280 } = quoted(view);
1281 let live = if view.child_soft_delete {
1282 live_predicate(view, true)
1283 } else {
1284 String::new()
1285 };
1286 let filter = filter_predicate(view, CHILD_ALIAS);
1287 let aggregate = aggregate_expr(view, CHILD_ALIAS);
1288 // Both sub-selects correlate on the parent, so the tenant check is a
1289 // plain column comparison inside each: no extra round trip, and a
1290 // cross-tenant child contributes to neither the count nor the row set.
1291 let tenant = tenant_predicate_joined(view);
1292 let accumulator = accumulator(view, &counter_column);
1293 format!(
1294 "UPDATE {parent_table} SET {counter_column} = {accumulator} - \
1295 (SELECT {aggregate} FROM {child_table} AS {CHILD_ALIAS} \
1296 WHERE {CHILD_ALIAS}.{fk_column} = {parent_table}.{parent_pk} \
1297 AND {CHILD_ALIAS}.{child_pk} IN ({id_list}){live}{filter}{tenant}) \
1298 WHERE {parent_table}.{parent_pk} IN \
1299 (SELECT {CHILD_ALIAS}.{fk_column} FROM {child_table} AS {CHILD_ALIAS} \
1300 WHERE {CHILD_ALIAS}.{child_pk} IN ({id_list}) \
1301 AND {CHILD_ALIAS}.{fk_column} IS NOT NULL{live}{filter} \
1302 AND {CHILD_ALIAS}.{fk_column} = {parent_table}.{parent_pk}{tenant})"
1303 )
1304}
1305
1306/// Render `ids` as a literal SQL list.
1307///
1308/// The values are `i64`, so their decimal rendering cannot contain anything a
1309/// SQL parser would treat as syntax — this is a type-level guarantee, not an
1310/// escaping convention. Inlining them (rather than binding) keeps the statement
1311/// backend-portable: `SQLite` has no array type for a bound `= ANY($1)`.
1312fn id_list(ids: &[i64]) -> String {
1313 let mut out = String::new();
1314 for (i, id) in ids.iter().enumerate() {
1315 if i > 0 {
1316 out.push(',');
1317 }
1318 out.push_str(&id.to_string());
1319 }
1320 out
1321}
1322
1323/// Increment every counter-cached parent of a child being **restored** from a
1324/// soft delete.
1325///
1326/// Must be called **before** `deleted_at` is cleared: the delta only applies
1327/// while the row is still soft-deleted, so restoring an already-live row is a
1328/// no-op rather than an inflation.
1329///
1330/// # Errors
1331///
1332/// Propagates any database error from the `UPDATE`s.
1333#[doc(hidden)]
1334pub async fn counter_cache_before_restore_by_id<M: 'static>(
1335 conn: &mut RuntimeConnection,
1336 specs: &[CounterCacheSpec<M>],
1337 child_id: i64,
1338) -> AutumnResult<()> {
1339 counter_cache_serialize_self_referential(conn, specs).await?;
1340 for index in specs_in_lock_order(specs) {
1341 let spec = &specs[index];
1342 // A model with no `deleted_at` has no restore path; guard anyway so a
1343 // mis-wired call cannot double-increment.
1344 if spec.child_soft_delete {
1345 counter_cache_apply_delta_by_child_id(conn, spec, child_id, 1, ChildState::SoftDeleted)
1346 .await?;
1347 }
1348 }
1349 Ok(())
1350}
1351
1352/// Read each counter-cached leg's current foreign key **and contribution** for
1353/// child `child_id`.
1354///
1355/// Called **before** an update so the post-update record can be compared against
1356/// it; the outer `Option` is `None` when the child row does not exist, has no
1357/// parent, or is soft-deleted. Issues no statement at all when the model has no
1358/// counter caches.
1359///
1360/// The contribution is read here rather than recomputed later because the
1361/// pre-update row is what it is a function of, and that row is gone once the
1362/// `UPDATE` lands. A row the filter rejects reports `0`, which is what makes a
1363/// filter flip on an unchanged parent a `+1` rather than a no-op.
1364///
1365/// # Errors
1366///
1367/// Propagates any database error from the `SELECT`s.
1368#[doc(hidden)]
1369pub async fn counter_cache_capture_fks<M: 'static>(
1370 conn: &mut RuntimeConnection,
1371 specs: &[CounterCacheSpec<M>],
1372 child_id: i64,
1373) -> AutumnResult<Vec<CapturedContribution>> {
1374 counter_cache_serialize_self_referential(conn, specs).await?;
1375 if specs.is_empty() {
1376 return Ok(Vec::new());
1377 }
1378 let mut out = Vec::with_capacity(specs.len());
1379 for spec in specs {
1380 let view = view(spec);
1381 assert_spec_idents(&view);
1382 let Quoted {
1383 child_table,
1384 child_pk,
1385 fk_column,
1386 ..
1387 } = quoted(&view);
1388 // A soft-deleted child is counted by nobody, so it has no "old parent"
1389 // to move away from — reporting one would make a later re-parent
1390 // decrement a counter that had already dropped this row.
1391 let live = live_predicate(&view, true);
1392 // A leg that reads the tenant captures it as well, as text: the old
1393 // side of a tenant change is guarded by this value, not by the row's
1394 // (by then new) tenant.
1395 if let Some(tenant_column) = captured_tenant_column(spec, &view) {
1396 let contrib = contrib_case_expr(&view, CHILD_ALIAS);
1397 let sql = format!(
1398 "SELECT {CHILD_ALIAS}.{fk_column} AS fk_value, \
1399 {contrib} AS contrib_value, \
1400 CAST({CHILD_ALIAS}.{tenant_column} AS TEXT) AS tenant_text \
1401 FROM {child_table} AS {CHILD_ALIAS} \
1402 WHERE {CHILD_ALIAS}.{child_pk} = {PH1}{live}{FOR_UPDATE}"
1403 );
1404 let row: Option<FkContribTenantRow> = diesel::sql_query(sql)
1405 .bind::<BigInt, _>(child_id)
1406 .get_result::<FkContribTenantRow>(conn)
1407 .await
1408 .optional_row()?;
1409 out.push(row.and_then(FkContribTenantRow::captured));
1410 continue;
1411 }
1412 if view.is_plain() {
1413 let sql = format!(
1414 "SELECT {CHILD_ALIAS}.{fk_column} AS fk_value \
1415 FROM {child_table} AS {CHILD_ALIAS} \
1416 WHERE {CHILD_ALIAS}.{child_pk} = {PH1}{live}{FOR_UPDATE}"
1417 );
1418 let row: Option<FkRow> = diesel::sql_query(sql)
1419 .bind::<BigInt, _>(child_id)
1420 .get_result::<FkRow>(conn)
1421 .await
1422 .optional_row()?;
1423 out.push(row.and_then(|r| r.fk_value).map(|fk| captured(fk, 1)));
1424 continue;
1425 }
1426 let contrib = contrib_case_expr(&view, CHILD_ALIAS);
1427 let sql = format!(
1428 "SELECT {CHILD_ALIAS}.{fk_column} AS fk_value, \
1429 {contrib} AS contrib_value \
1430 FROM {child_table} AS {CHILD_ALIAS} \
1431 WHERE {CHILD_ALIAS}.{child_pk} = {PH1}{live}{FOR_UPDATE}"
1432 );
1433 let row: Option<FkContribRow> = diesel::sql_query(sql)
1434 .bind::<BigInt, _>(child_id)
1435 .get_result::<FkContribRow>(conn)
1436 .await
1437 .optional_row()?;
1438 out.push(row.and_then(|r| r.fk_value.map(|fk| captured(fk, r.contrib_value))));
1439 }
1440 Ok(out)
1441}
1442
1443/// A [`Captured`] with no tenant, for the legs that do not read one.
1444const fn captured(parent: i64, contrib: i64) -> Captured {
1445 Captured {
1446 parent,
1447 contrib,
1448 tenant: None,
1449 }
1450}
1451
1452/// The quoted tenant column the capture reads, when this leg reads one: it
1453/// has a tenant column *and* an accessor for the new side to compare against.
1454/// Without the accessor the capture stays as it was, so a leg that cannot see
1455/// its tenant costs nothing extra.
1456fn captured_tenant_column<M: 'static>(
1457 spec: &CounterCacheSpec<M>,
1458 view: &SqlView,
1459) -> Option<String> {
1460 let tenant_column = view.tenant_column?;
1461 spec.tenant_of?;
1462 Some(quote_ident(tenant_column))
1463}
1464
1465/// Move the counters for a child whose foreign keys may have just changed.
1466///
1467/// `before` is the slice [`counter_cache_capture_fks`] returned prior to the
1468/// update; `record` is the row as persisted after it. A leg whose foreign key is
1469/// unchanged issues **no** statement — an ordinary field edit costs nothing.
1470///
1471/// # Errors
1472///
1473/// Propagates any database error from the `UPDATE`s.
1474#[doc(hidden)]
1475pub async fn counter_cache_after_update<M: Send + Sync + 'static>(
1476 conn: &mut RuntimeConnection,
1477 specs: &[CounterCacheSpec<M>],
1478 before: &[CapturedContribution],
1479 record: &M,
1480) -> AutumnResult<()> {
1481 let mut moves: Vec<Contribution> = Vec::with_capacity(specs.len() * 2);
1482 for (index, spec) in specs.iter().enumerate() {
1483 let old = before.get(index).cloned().flatten();
1484 let new = contribution_of(spec, record);
1485 // Collected rather than applied here, and not old-then-new: every delta
1486 // this mutation makes goes out in one global lock order (see
1487 // `apply_ordered`), so two transactions swapping children between the
1488 // same parents cannot take the two row locks in opposite orders.
1489 push_diff(index, old, new, (spec.pk_of)(record), &mut moves);
1490 }
1491 apply_ordered(conn, specs, moves).await
1492}
1493
1494/// A record's `(parent, contribution)` after a mutation, or `None` when it
1495/// contributes to no parent.
1496///
1497/// A soft-deleted child is counted by nobody, so neither its old nor its new
1498/// parent may move. The generated `update` does not filter soft-deleted rows, so
1499/// this is reachable.
1500fn contribution_of<M: 'static>(spec: &CounterCacheSpec<M>, record: &M) -> CapturedContribution {
1501 if !(spec.live_of)(record) {
1502 return None;
1503 }
1504 (spec.fk_of)(record).map(|parent| Captured {
1505 parent,
1506 contrib: (spec.contrib_of)(record),
1507 tenant: spec.tenant_of.and_then(|tenant_of| tenant_of(record)),
1508 })
1509}
1510
1511/// Turn one leg's before/after [`Captured`] into deltas.
1512///
1513/// Unchanged ⇒ nothing. Same parent and tenant, different weight (a filter
1514/// flip, or an edited summed field) ⇒ one delta for the difference. Different
1515/// parent, or different tenant ⇒ the old weight off the old parent and the new
1516/// weight onto the new one, each skipped when it is 0, so a row the filter
1517/// rejects never touches a parent row.
1518///
1519/// The new side is always scoped by the child row as it now is. The old side
1520/// is too, unless the tenant changed: then the row no longer says which tenant
1521/// the old parent sits in, so the removal is scoped by the tenant the capture
1522/// read, and a parent left behind in the old tenant is found rather than
1523/// missed.
1524fn push_diff(
1525 index: usize,
1526 old: CapturedContribution,
1527 new: CapturedContribution,
1528 witness: i64,
1529 out: &mut Vec<Contribution>,
1530) {
1531 if old == new {
1532 return;
1533 }
1534 let live = TenantScope::SameTenantAsChild(witness);
1535 let new_tenant = new.as_ref().and_then(|new| new.tenant.clone());
1536 if let (Some(old), Some(new)) = (&old, &new)
1537 && old.parent == new.parent
1538 && old.tenant == new.tenant
1539 {
1540 // Two representable weights need not have a representable difference
1541 // (`i64::MIN` edited to `i64::MAX`). Then the move goes out as the old
1542 // weight off and the new weight on, which reach the same total.
1543 if let Some(delta) = new.contrib.checked_sub(old.contrib) {
1544 if delta != 0 {
1545 out.push((index, old.parent, delta, live));
1546 }
1547 return;
1548 }
1549 push_removal(index, old.parent, old.contrib, live.clone(), out);
1550 out.push((index, new.parent, new.contrib, live));
1551 return;
1552 }
1553 if let Some(old) = old
1554 && old.contrib != 0
1555 {
1556 let scope = if old.tenant == new_tenant {
1557 live.clone()
1558 } else {
1559 TenantScope::Captured(old.tenant)
1560 };
1561 push_removal(index, old.parent, old.contrib, scope, out);
1562 }
1563 if let Some(new) = new
1564 && new.contrib != 0
1565 {
1566 out.push((index, new.parent, new.contrib, live));
1567 }
1568}
1569
1570/// Take `contrib` off `parent`.
1571///
1572/// `-i64::MIN` is not an `i64`, so that one weight leaves as two halves that
1573/// are. Every other weight negates in place.
1574fn push_removal(
1575 index: usize,
1576 parent: i64,
1577 contrib: i64,
1578 scope: TenantScope,
1579 out: &mut Vec<Contribution>,
1580) {
1581 if let Some(negated) = contrib.checked_neg() {
1582 out.push((index, parent, negated, scope));
1583 } else {
1584 let half = contrib / 2;
1585 out.push((index, parent, -half, scope.clone()));
1586 out.push((index, parent, -(contrib - half), scope));
1587 }
1588}
1589
1590/// [`counter_cache_capture_fks`] over a batch of child ids (`update_many`).
1591///
1592/// Issues one `SELECT` per spec (not per id), returning `(child id,
1593/// (parent, contribution) in spec order)` for every row found. Rows absent from
1594/// the table are simply absent from the result.
1595///
1596/// # Errors
1597///
1598/// Propagates any database error from the `SELECT`s.
1599#[doc(hidden)]
1600pub async fn counter_cache_capture_fks_many<M: 'static>(
1601 conn: &mut RuntimeConnection,
1602 specs: &[CounterCacheSpec<M>],
1603 child_ids: &[i64],
1604) -> AutumnResult<CapturedContributions> {
1605 counter_cache_serialize_self_referential(conn, specs).await?;
1606 if specs.is_empty() || child_ids.is_empty() {
1607 return Ok(Vec::new());
1608 }
1609 let id_list = id_list(child_ids);
1610 let mut by_child: HashMap<i64, Vec<CapturedContribution>> = HashMap::new();
1611 for (index, spec) in specs.iter().enumerate() {
1612 let view = view(spec);
1613 assert_spec_idents(&view);
1614 let Quoted {
1615 child_table,
1616 child_pk,
1617 fk_column,
1618 ..
1619 } = quoted(&view);
1620 let live = live_predicate(&view, true);
1621 if let Some(tenant_column) = captured_tenant_column(spec, &view) {
1622 let contrib = contrib_case_expr(&view, CHILD_ALIAS);
1623 let sql = format!(
1624 "SELECT {CHILD_ALIAS}.{child_pk} AS child_id, \
1625 {CHILD_ALIAS}.{fk_column} AS fk_value, \
1626 {contrib} AS contrib_value, \
1627 CAST({CHILD_ALIAS}.{tenant_column} AS TEXT) AS tenant_text \
1628 FROM {child_table} AS {CHILD_ALIAS} \
1629 WHERE {CHILD_ALIAS}.{child_pk} IN ({id_list}){live} \
1630 ORDER BY {CHILD_ALIAS}.{child_pk}{FOR_UPDATE}"
1631 );
1632 let rows: Vec<ChildFkContribTenantRow> = diesel::sql_query(sql)
1633 .load::<ChildFkContribTenantRow>(conn)
1634 .await
1635 .map_err(AutumnError::from)?;
1636 for row in rows {
1637 let entry = by_child
1638 .entry(row.child_id)
1639 .or_insert_with(|| vec![None; specs.len()]);
1640 entry[index] = row.captured();
1641 }
1642 continue;
1643 }
1644 if view.is_plain() {
1645 let sql = format!(
1646 "SELECT {CHILD_ALIAS}.{child_pk} AS child_id, \
1647 {CHILD_ALIAS}.{fk_column} AS fk_value \
1648 FROM {child_table} AS {CHILD_ALIAS} \
1649 WHERE {CHILD_ALIAS}.{child_pk} IN ({id_list}){live} \
1650 ORDER BY {CHILD_ALIAS}.{child_pk}{FOR_UPDATE}"
1651 );
1652 let rows: Vec<ChildFkRow> = diesel::sql_query(sql)
1653 .load::<ChildFkRow>(conn)
1654 .await
1655 .map_err(AutumnError::from)?;
1656 for row in rows {
1657 let entry = by_child
1658 .entry(row.child_id)
1659 .or_insert_with(|| vec![None; specs.len()]);
1660 entry[index] = row.fk_value.map(|fk| captured(fk, 1));
1661 }
1662 continue;
1663 }
1664 let contrib = contrib_case_expr(&view, CHILD_ALIAS);
1665 let sql = format!(
1666 "SELECT {CHILD_ALIAS}.{child_pk} AS child_id, \
1667 {CHILD_ALIAS}.{fk_column} AS fk_value, \
1668 {contrib} AS contrib_value \
1669 FROM {child_table} AS {CHILD_ALIAS} \
1670 WHERE {CHILD_ALIAS}.{child_pk} IN ({id_list}){live} \
1671 ORDER BY {CHILD_ALIAS}.{child_pk}{FOR_UPDATE}"
1672 );
1673 let rows: Vec<ChildFkContribRow> = diesel::sql_query(sql)
1674 .load::<ChildFkContribRow>(conn)
1675 .await
1676 .map_err(AutumnError::from)?;
1677 for row in rows {
1678 let entry = by_child
1679 .entry(row.child_id)
1680 .or_insert_with(|| vec![None; specs.len()]);
1681 entry[index] = row.fk_value.map(|fk| captured(fk, row.contrib_value));
1682 }
1683 }
1684 let mut out: CapturedContributions = by_child.into_iter().collect();
1685 out.sort_unstable_by_key(|(id, _)| *id);
1686 Ok(out)
1687}
1688
1689/// [`counter_cache_after_update`] over a batch of post-update records
1690/// (`update_many`), matched back to their pre-update foreign keys by primary key.
1691///
1692/// A record with no captured "before" entry (its row appeared between the
1693/// capture and the update) is skipped rather than guessed at; the recompute
1694/// repair path exists for exactly that kind of edge.
1695///
1696/// # Errors
1697///
1698/// Propagates any database error from the `UPDATE`s.
1699#[doc(hidden)]
1700pub async fn counter_cache_after_update_many<M: Send + Sync + 'static>(
1701 conn: &mut RuntimeConnection,
1702 specs: &[CounterCacheSpec<M>],
1703 before: &[(i64, Vec<CapturedContribution>)],
1704 records: &[M],
1705) -> AutumnResult<()> {
1706 if specs.is_empty() {
1707 return Ok(());
1708 }
1709 let pk_of = specs[0].pk_of;
1710 let mut contributions: Vec<Contribution> = Vec::new();
1711 for (index, spec) in specs.iter().enumerate() {
1712 // Every leg's deltas are collected before any statement runs, then
1713 // folded and applied in one global lock order (see `apply_ordered`).
1714 // Re-parenting 1000 children from one post to another is then two
1715 // `UPDATE`s (`-1000`, `+1000`) rather than 2000, and two concurrent
1716 // batches swapping children between the same parents take their row
1717 // locks in the same order and cannot deadlock.
1718 for record in records {
1719 let child_id = pk_of(record);
1720 let Ok(found) = before.binary_search_by_key(&child_id, |(id, _)| *id) else {
1721 // No captured "before" for this row: it appeared between the
1722 // capture and the update. Guessing would be worse than leaving
1723 // it to `recompute`.
1724 continue;
1725 };
1726 let old = before[found].1.get(index).cloned().flatten();
1727 let new = contribution_of(spec, record);
1728 push_diff(index, old, new, child_id, &mut contributions);
1729 }
1730 }
1731 apply_ordered(conn, specs, contributions).await
1732}
1733
1734/// Move the counters for an `upsert_many` chunk (#1325).
1735///
1736/// `existing` is the pre-upsert snapshot the upsert already loaded (and locked)
1737/// for this chunk, so no extra query is needed: a record present in it is an
1738/// **update** and gets the before/after foreign-key diff; a record absent from it
1739/// is an **insert** and gets a plain increment.
1740///
1741/// # The one row this classification can get wrong
1742///
1743/// The snapshot is loaded `FOR UPDATE`, so every row that *existed* when it ran
1744/// is locked: it cannot be re-parented underneath this chunk, and its diff is
1745/// exact. The gap is the row that did **not** exist then — there is nothing to
1746/// lock — and that another transaction inserts before this chunk's `INSERT …
1747/// ON CONFLICT` runs. Postgres then updates that row rather than inserting it,
1748/// while this classifies it as an insert: the `+1` duplicates the increment the
1749/// other transaction already made, and if the upsert also moved the foreign key,
1750/// the old parent never gets its decrement.
1751///
1752/// Closing it needs the discrimination to come from the statement that did the
1753/// work (Postgres exposes it as `xmax = 0` in `RETURNING`), because no
1754/// pre-upsert read can see a row that does not exist yet — row locks do not
1755/// block inserts, and the advisory lock the *versioned* upsert takes only
1756/// serializes upserts against each other, not against a plain `save` or a raw
1757/// insert. It is drift, not corruption, and `recompute` repairs it.
1758///
1759/// `SQLite` is unaffected: `BEGIN IMMEDIATE` excludes every other writer for the
1760/// duration of the transaction, so the window does not exist there.
1761///
1762/// # Errors
1763///
1764/// Propagates any database error from the `UPDATE`s.
1765#[doc(hidden)]
1766pub async fn counter_cache_after_upsert_many<M: Send + Sync + 'static>(
1767 conn: &mut RuntimeConnection,
1768 specs: &[CounterCacheSpec<M>],
1769 existing: &[M],
1770 upserted: &[M],
1771) -> AutumnResult<()> {
1772 counter_cache_serialize_self_referential(conn, specs).await?;
1773 if specs.is_empty() {
1774 return Ok(());
1775 }
1776 let pk_of = specs[0].pk_of;
1777 // A soft-deleted existing row is counted by nobody, so it has no "old
1778 // parent". Recording one would make the post-upsert side — which now checks
1779 // `live_of` and yields `None` for a row that stays deleted — decrement a
1780 // parent that had already dropped it.
1781 let before: HashMap<i64, Vec<CapturedContribution>> = existing
1782 .iter()
1783 .map(|row| {
1784 (
1785 pk_of(row),
1786 specs
1787 .iter()
1788 .map(|spec| contribution_of(spec, row))
1789 .collect(),
1790 )
1791 })
1792 .collect();
1793 // One pass per leg over the whole chunk, inserts and updates together: a row
1794 // absent from `before` is an insert (`+1`), a row present in it is an update
1795 // (the before/after diff). Folded and ordered by `apply_ordered` like every
1796 // other bulk path, for the same two reasons.
1797 let mut contributions: Vec<Contribution> = Vec::new();
1798 for (index, spec) in specs.iter().enumerate() {
1799 for record in upserted {
1800 let child_id = pk_of(record);
1801 let new = contribution_of(spec, record);
1802 match before.get(&child_id) {
1803 None => {
1804 if let Some(new) = new
1805 && new.contrib != 0
1806 {
1807 let scope = TenantScope::SameTenantAsChild(child_id);
1808 contributions.push((index, new.parent, new.contrib, scope));
1809 }
1810 }
1811 Some(old_fks) => {
1812 let old = old_fks.get(index).cloned().flatten();
1813 push_diff(index, old, new, child_id, &mut contributions);
1814 }
1815 }
1816 }
1817 }
1818 apply_ordered(conn, specs, contributions).await
1819}
1820
1821/// How many parent rows one recompute transaction locks and repairs at a time.
1822///
1823/// A full sweep has to take a row lock on every parent it rebuilds (see
1824/// [`recompute_batch`]), and locks are held until commit. Rebuilding the whole
1825/// table in one transaction would therefore block every concurrent write to that
1826/// table for the duration of the sweep, so the sweep is cut into batches: each
1827/// one is its own short transaction, and per-parent correctness does not depend
1828/// on the batches being atomic with each other.
1829const RECOMPUTE_BATCH: i64 = 1_000;
1830
1831/// The maintained value as the source of truth defines it, correlated on the
1832/// parent row.
1833///
1834/// The correlated sub-select aliases the child table so a self-referential
1835/// counter cache (child table == parent table) still binds the outer statement's
1836/// target on the right-hand side of the join predicate.
1837///
1838/// The ground truth has to agree with what the deltas maintain: an ordinary
1839/// delta skips a cross-tenant child and a filtered-out row, so a repair that
1840/// counted either would undo the isolation, or the filter, on the very next
1841/// sweep.
1842pub fn ground_truth_sql(view: &SqlView) -> String {
1843 let Quoted {
1844 child_table,
1845 fk_column,
1846 parent_table,
1847 parent_pk,
1848 ..
1849 } = quoted(view);
1850 let live = live_predicate(view, true);
1851 let filter = filter_predicate(view, CHILD_ALIAS);
1852 let tenant = tenant_predicate_joined(view);
1853 let aggregate = aggregate_expr(view, CHILD_ALIAS);
1854 format!(
1855 "(SELECT {aggregate} FROM {child_table} AS {CHILD_ALIAS} \
1856 WHERE {CHILD_ALIAS}.{fk_column} = {parent_table}.{parent_pk}{live}{filter}{tenant})"
1857 )
1858}
1859
1860/// The `UPDATE` that rebuilds `ids`' maintained values from the source of truth.
1861pub fn recompute_update_sql(view: &SqlView, ids: &str) -> String {
1862 let Quoted {
1863 parent_table,
1864 parent_pk,
1865 counter_column,
1866 ..
1867 } = quoted(view);
1868 let ground_truth = ground_truth_sql(view);
1869 // `IS DISTINCT FROM` so a sweep over a healthy table writes nothing: under
1870 // MVCC an unconditional assignment would rewrite every parent row (bloat
1871 // proportional to the whole table, for no change), and it would make the
1872 // returned count the row count rather than the repair count.
1873 format!(
1874 "UPDATE {parent_table} SET {counter_column} = {ground_truth} \
1875 WHERE {parent_table}.{parent_pk} IN ({ids}) \
1876 AND {parent_table}.{counter_column} {IS_DISTINCT_FROM} {ground_truth}"
1877 )
1878}
1879
1880/// How many of the newest `limit` parent rows disagree with the source of
1881/// truth.
1882///
1883/// One aggregate statement per derivation, so it is a single round trip. The
1884/// `LIMIT` bounds the **input**: the `limit` highest-id parents are the window,
1885/// materialised under the parent table's own name so the correlated ground
1886/// truth needs no rewriting, and only those rows pay for a correlated
1887/// aggregate. A `LIMIT` on the matching output alone would not do it: on a
1888/// healthy table nothing matches, so the engine would evaluate the aggregate
1889/// for every parent before proving there is nothing left to find. The question
1890/// an operator asks is "is this derivation drifting", not "by exactly how much
1891/// on a billion-row table", and newest-first is where live writes, and so live
1892/// drift, concentrate.
1893pub fn drift_sql(view: &SqlView, limit: i64) -> String {
1894 let Quoted {
1895 parent_table,
1896 parent_pk,
1897 counter_column,
1898 ..
1899 } = quoted(view);
1900 let ground_truth = ground_truth_sql(view);
1901 format!(
1902 "SELECT COUNT(*) AS count FROM \
1903 (SELECT 1 AS drifted FROM \
1904 (SELECT * FROM {parent_table} ORDER BY {parent_table}.{parent_pk} DESC LIMIT {limit}) \
1905 AS {parent_table} \
1906 WHERE {parent_table}.{counter_column} {IS_DISTINCT_FROM} {ground_truth}) \
1907 AS __autumn_cc_drift"
1908 )
1909}
1910
1911/// Repair one batch of parents: lock them, then rebuild them, in one transaction.
1912///
1913/// The lock is what makes a repair safe to run against live traffic. Without it,
1914/// on Postgres, a repair racing an in-flight child insert **introduces** the
1915/// drift it exists to remove: the child's transaction has already taken the
1916/// parent's row lock for its `SET c = c + 1` but has not committed, so the
1917/// repair's `UPDATE` blocks on that lock, and when it resumes it writes a
1918/// `COUNT(*)` taken from a snapshot that predates — and therefore cannot see —
1919/// the child that just committed. The committed `+1` is silently overwritten.
1920///
1921/// Taking the lock in a *separate, earlier statement* removes both halves of
1922/// that race. Either the repair wins the lock, in which case the child's
1923/// increment is applied after the repair commits and is relative, so it lands on
1924/// top of the rebuilt value; or the child wins, in which case the locking
1925/// `SELECT` waits for it to commit and the `UPDATE` that follows takes a fresh
1926/// snapshot that does see it. Ids are locked in ascending order, matching the
1927/// order the delta paths apply in, so the two cannot deadlock against each
1928/// other.
1929///
1930/// On `SQLite` the enclosing `BEGIN IMMEDIATE` already excludes every other
1931/// writer, and `FOR UPDATE` degrades to the empty string; the locking `SELECT`
1932/// is then a cheap indexed read that keeps this one code path.
1933async fn recompute_batch(
1934 conn: &mut RuntimeConnection,
1935 view: &SqlView,
1936 ids: &[i64],
1937) -> AutumnResult<usize> {
1938 if ids.is_empty() {
1939 return Ok(0);
1940 }
1941 let view = *view;
1942 let ids = ids.to_vec();
1943 scoped_immediate_transaction::<usize, AutumnError, _>(conn, move |conn| {
1944 async move { recompute_batch_statements(conn, &view, &ids).await }.scope_boxed()
1945 })
1946 .await
1947}
1948
1949/// The two statements [`recompute_batch`] runs, without the transaction.
1950///
1951/// Split out because the derivation backfill has to commit the repaired batch
1952/// and its checkpoint **together**: it opens one transaction and runs this plus
1953/// the checkpoint write inside it. A checkpoint committed separately from the
1954/// batch it describes would double-apply or skip a batch after a crash.
1955pub async fn recompute_batch_statements(
1956 conn: &mut RuntimeConnection,
1957 view: &SqlView,
1958 ids: &[i64],
1959) -> AutumnResult<usize> {
1960 if ids.is_empty() {
1961 return Ok(0);
1962 }
1963 assert_spec_idents(view);
1964 let id_list = id_list(ids);
1965 let parent_table = quote_ident(view.parent_table);
1966 let parent_pk = quote_ident(view.parent_pk);
1967 let lock_sql = format!(
1968 "SELECT {parent_pk} AS id FROM {parent_table} \
1969 WHERE {parent_pk} IN ({id_list}) ORDER BY {parent_pk}{FOR_UPDATE}"
1970 );
1971 diesel::sql_query(lock_sql)
1972 .load::<IdRow>(&mut *conn)
1973 .await
1974 .map_err(AutumnError::from)?;
1975 diesel::sql_query(recompute_update_sql(view, &id_list))
1976 .execute(&mut *conn)
1977 .await
1978 .map_err(AutumnError::from)
1979}
1980
1981/// One page of parent primary keys after `cursor`, in ascending order.
1982///
1983/// A parent inserted after its page was read is simply not in this sweep, which
1984/// is harmless: it starts at the column default and its children are counted by
1985/// the delta paths.
1986///
1987/// [`recompute_view`] reads pages outside its repair transactions, so no lock is
1988/// held while enumerating. The derivation backfill
1989/// ([`crate::derivation::run_backfill`]) reads its page **inside** the batch
1990/// transaction instead, because the cursor it pages from is the checkpoint on
1991/// the state row that transaction holds the lock on. Reading it outside would
1992/// mean paging from a cursor another replica had already moved.
1993pub async fn parent_id_page(
1994 conn: &mut RuntimeConnection,
1995 view: &SqlView,
1996 cursor: Option<i64>,
1997 limit: i64,
1998) -> AutumnResult<Vec<i64>> {
1999 assert_spec_idents(view);
2000 let parent_table = quote_ident(view.parent_table);
2001 let parent_pk = quote_ident(view.parent_pk);
2002 let mut page_sql = format!("SELECT {parent_pk} AS id FROM {parent_table}");
2003 if cursor.is_some() {
2004 let _ = write!(page_sql, " WHERE {parent_pk} > {PH1}");
2005 }
2006 let _ = write!(page_sql, " ORDER BY {parent_pk} LIMIT {limit}");
2007 let query = diesel::sql_query(page_sql);
2008 let page = if let Some(after) = cursor {
2009 query.bind::<BigInt, _>(after).load::<IdRow>(conn).await
2010 } else {
2011 query.load::<IdRow>(conn).await
2012 }
2013 .map_err(AutumnError::from)?;
2014 Ok(page.iter().map(|row| row.id).collect())
2015}
2016
2017/// Rebuild one maintained value from the source of truth, one batch at a time.
2018///
2019/// The non-generic core of [`counter_cache_recompute`], so the derivation
2020/// repair path ([`crate::derivation::recompute`]) runs exactly the same sweep.
2021pub async fn recompute_view(
2022 conn: &mut RuntimeConnection,
2023 view: &SqlView,
2024 parent_id: Option<i64>,
2025) -> AutumnResult<usize> {
2026 assert_spec_idents(view);
2027 if let Some(id) = parent_id {
2028 return recompute_batch(conn, view, &[id]).await;
2029 }
2030 // A self-referential table (a comment's `reply_count`) has rows that are
2031 // children and parents at once: a batch holding several parents in id
2032 // order could form a lock cycle with a mutation that holds a child row and
2033 // wants its parent. One parent per transaction takes one lock and never a
2034 // second, so the repair cannot be one side of that cycle.
2035 let batch = if view.child_table == view.parent_table {
2036 1
2037 } else {
2038 RECOMPUTE_BATCH
2039 };
2040 let mut touched = 0usize;
2041 let mut cursor: Option<i64> = None;
2042 loop {
2043 let ids = parent_id_page(conn, view, cursor, batch).await?;
2044 let Some(&last) = ids.last() else { break };
2045 cursor = Some(last);
2046 touched += recompute_batch_retrying(conn, view, &ids).await?;
2047 }
2048 Ok(touched)
2049}
2050
2051/// How many times a repair batch that lost a lock race is retried before the
2052/// error surfaces. Each batch is its own transaction, so a retry repeats no
2053/// committed work.
2054const LOCK_CONTENTION_RETRIES: u32 = 5;
2055
2056/// [`recompute_batch`], retried when the database aborted the batch to break a
2057/// lock cycle or a serialisation conflict with a concurrent writer.
2058async fn recompute_batch_retrying(
2059 conn: &mut RuntimeConnection,
2060 view: &SqlView,
2061 ids: &[i64],
2062) -> AutumnResult<usize> {
2063 let mut attempt = 0;
2064 loop {
2065 match recompute_batch(conn, view, ids).await {
2066 Err(error) if attempt < LOCK_CONTENTION_RETRIES && is_lock_contention(&error) => {
2067 attempt += 1;
2068 tracing::warn!(
2069 parent_table = view.parent_table,
2070 column = view.counter_column,
2071 attempt,
2072 error = %error,
2073 "repair batch lost a lock race; retrying"
2074 );
2075 tokio::time::sleep(std::time::Duration::from_millis(25 * u64::from(attempt))).await;
2076 }
2077 outcome => return outcome,
2078 }
2079 }
2080}
2081
2082/// Whether `error` is the database aborting one transaction so that another
2083/// can proceed: Postgres's deadlock detector (`40P01`) or a serialisation
2084/// failure (`40001`), and `SQLite`'s busy timeout.
2085pub fn is_lock_contention(error: &AutumnError) -> bool {
2086 let message = error.to_string();
2087 message.contains("deadlock detected")
2088 || message.contains("could not serialize access")
2089 || message.contains("database is locked")
2090}
2091
2092/// Recompute counters from the source of truth.
2093///
2094/// With `parent_id = None` every parent row is rebuilt; with `Some(id)` only
2095/// that parent is touched. Idempotent by construction — the column is *assigned*
2096/// a `COUNT(*)`, never adjusted — so it is safe to run repeatedly, and it is the
2097/// supported way to adopt a counter column on an existing table (AC6).
2098///
2099/// Safe to run against live traffic: each batch locks the parents it is about to
2100/// rebuild before counting, so it can neither read a half-applied write nor
2101/// clobber one (see [`recompute_batch`]).
2102///
2103/// Returns the number of parent rows **actually repaired**, summed across every
2104/// spec — a sweep over a table with no drift returns 0 and writes nothing.
2105///
2106/// # Errors
2107///
2108/// Propagates any database error from the `SELECT`s or `UPDATE`s.
2109#[doc(hidden)]
2110pub async fn counter_cache_recompute<M: 'static>(
2111 conn: &mut RuntimeConnection,
2112 specs: &[CounterCacheSpec<M>],
2113 parent_id: Option<i64>,
2114) -> AutumnResult<usize> {
2115 let mut touched = 0usize;
2116 for index in specs_in_lock_order(specs) {
2117 touched += recompute_view(conn, &view(&specs[index]), parent_id).await?;
2118 }
2119 Ok(touched)
2120}
2121
2122#[derive(diesel::QueryableByName)]
2123struct ChildFkRow {
2124 #[diesel(sql_type = BigInt)]
2125 child_id: i64,
2126 #[diesel(sql_type = Nullable<BigInt>)]
2127 fk_value: Option<i64>,
2128}
2129
2130/// [`ChildFkRow`] plus the row's contribution, for a derivation's bulk capture.
2131#[derive(diesel::QueryableByName)]
2132struct ChildFkContribRow {
2133 #[diesel(sql_type = BigInt)]
2134 child_id: i64,
2135 #[diesel(sql_type = Nullable<BigInt>)]
2136 fk_value: Option<i64>,
2137 #[diesel(sql_type = BigInt)]
2138 contrib_value: i64,
2139}
2140
2141/// A single `i64` primary key, aliased `id`.
2142///
2143/// Shared by the two statements whose result is incidental: the recompute
2144/// sweep's page of parent ids, and the bulk-delete child lock (whose rows are
2145/// discarded — the point is the lock it takes).
2146#[derive(diesel::QueryableByName)]
2147struct IdRow {
2148 #[diesel(sql_type = BigInt)]
2149 id: i64,
2150}
2151
2152#[derive(diesel::QueryableByName)]
2153struct FkRow {
2154 #[diesel(sql_type = Nullable<BigInt>)]
2155 fk_value: Option<i64>,
2156}
2157
2158/// [`FkRow`] plus the row's contribution, for a derivation's single capture.
2159#[derive(diesel::QueryableByName)]
2160struct FkContribRow {
2161 #[diesel(sql_type = Nullable<BigInt>)]
2162 fk_value: Option<i64>,
2163 #[diesel(sql_type = BigInt)]
2164 contrib_value: i64,
2165}
2166
2167/// [`FkContribRow`] plus the row's tenant as text, for a leg that reads it.
2168#[derive(diesel::QueryableByName)]
2169struct FkContribTenantRow {
2170 #[diesel(sql_type = Nullable<BigInt>)]
2171 fk_value: Option<i64>,
2172 #[diesel(sql_type = BigInt)]
2173 contrib_value: i64,
2174 #[diesel(sql_type = Nullable<Text>)]
2175 tenant_text: Option<String>,
2176}
2177
2178impl FkContribTenantRow {
2179 fn captured(self) -> CapturedContribution {
2180 self.fk_value.map(|parent| Captured {
2181 parent,
2182 contrib: self.contrib_value,
2183 tenant: self.tenant_text,
2184 })
2185 }
2186}
2187
2188/// [`FkContribTenantRow`] keyed by child, for the bulk capture.
2189#[derive(diesel::QueryableByName)]
2190struct ChildFkContribTenantRow {
2191 #[diesel(sql_type = BigInt)]
2192 child_id: i64,
2193 #[diesel(sql_type = Nullable<BigInt>)]
2194 fk_value: Option<i64>,
2195 #[diesel(sql_type = BigInt)]
2196 contrib_value: i64,
2197 #[diesel(sql_type = Nullable<Text>)]
2198 tenant_text: Option<String>,
2199}
2200
2201impl ChildFkContribTenantRow {
2202 fn captured(self) -> CapturedContribution {
2203 self.fk_value.map(|parent| Captured {
2204 parent,
2205 contrib: self.contrib_value,
2206 tenant: self.tenant_text,
2207 })
2208 }
2209}
2210
2211/// `Result::optional`, spelled locally so this module does not have to pull
2212/// diesel's `OptionalExtension` into every call site's scope.
2213trait OptionalRow<T> {
2214 fn optional_row(self) -> AutumnResult<Option<T>>;
2215}
2216
2217impl<T> OptionalRow<T> for Result<T, diesel::result::Error> {
2218 fn optional_row(self) -> AutumnResult<Option<T>> {
2219 match self {
2220 Ok(value) => Ok(Some(value)),
2221 Err(diesel::result::Error::NotFound) => Ok(None),
2222 Err(e) => Err(AutumnError::from(e)),
2223 }
2224 }
2225}
2226
2227#[cfg(test)]
2228mod tests {
2229 use super::*;
2230
2231 struct Dummy;
2232
2233 fn spec(soft: bool) -> CounterCacheSpec<Dummy> {
2234 CounterCacheSpec {
2235 child_table: "comments",
2236 child_pk: "id",
2237 child_soft_delete: soft,
2238 fk_column: "post_id",
2239 parent_table: "posts",
2240 parent_pk: "id",
2241 counter_column: "comment_count",
2242 fk_of: |_| Some(1),
2243 pk_of: |_| 1,
2244 live_of: |_| true,
2245 tenant_column: None,
2246 tenant_of: None,
2247 contrib_of: |_| 1,
2248 contrib_sql: "1",
2249 filter_sql: "",
2250 derivation: None,
2251 }
2252 }
2253
2254 /// A filtered `sum(score)` derivation over the same tables.
2255 fn sum_spec() -> CounterCacheSpec<Dummy> {
2256 CounterCacheSpec {
2257 counter_column: "visible_score",
2258 contrib_sql: "{c}.\"score\"",
2259 filter_sql: " AND ({c}.\"published\" = TRUE)",
2260 ..spec(false)
2261 }
2262 }
2263
2264 #[test]
2265 fn plain_identifiers_are_accepted_and_sql_fragments_are_not() {
2266 assert!(is_plain_identifier("comment_count"));
2267 assert!(is_plain_identifier("_x9"));
2268 assert!(!is_plain_identifier(""));
2269 assert!(!is_plain_identifier("9lives"));
2270 assert!(!is_plain_identifier("comment_count; DROP TABLE posts"));
2271 assert!(!is_plain_identifier("comment count"));
2272 assert!(!is_plain_identifier("\"comment_count\""));
2273 }
2274
2275 #[test]
2276 fn the_live_predicate_is_emitted_only_for_a_soft_deleting_child() {
2277 assert_eq!(live_predicate(&view(&spec(false)), true), "");
2278 assert_eq!(
2279 live_predicate(&view(&spec(true)), true),
2280 format!(" AND {CHILD_ALIAS}.\"deleted_at\" IS NULL")
2281 );
2282 assert_eq!(
2283 live_predicate(&view(&spec(true)), false),
2284 format!(" AND {CHILD_ALIAS}.\"deleted_at\" IS NOT NULL")
2285 );
2286 }
2287
2288 #[test]
2289 fn a_bulk_decrement_locks_its_children_in_ascending_id_order() {
2290 // Deterministic order, and children before parents — the same direction
2291 // the single-row and update paths take. Inverting it here would let a
2292 // bulk delete deadlock against an `update` that re-parents one of the
2293 // same children.
2294 let sql = child_lock_sql(&view(&spec(false)), "3,1,2");
2295 assert!(
2296 sql.starts_with("SELECT \"id\" AS id FROM \"comments\""),
2297 "{sql}"
2298 );
2299 assert!(sql.contains("WHERE \"id\" IN (3,1,2)"), "{sql}");
2300 assert!(
2301 sql.contains(&format!("ORDER BY \"id\"{FOR_UPDATE}")),
2302 "the lock must be taken in a deterministic order: {sql}"
2303 );
2304 }
2305
2306 #[test]
2307 fn a_recompute_batch_is_scoped_to_the_ids_it_locked() {
2308 // Restricting the `UPDATE` to the batch is what makes the locking
2309 // `SELECT` that precedes it meaningful: a sweep-wide `UPDATE` would
2310 // touch parents this transaction never locked.
2311 let sql = recompute_update_sql(&view(&spec(false)), "1,2,3");
2312 assert!(sql.contains("\"posts\".\"id\" IN (1,2,3)"), "{sql}");
2313 assert!(
2314 sql.contains(&format!("\"posts\".\"comment_count\" {IS_DISTINCT_FROM}")),
2315 "a healthy parent must still be left unwritten: {sql}"
2316 );
2317 }
2318
2319 /// Two legs whose parents sit in tables named in the *opposite* order to
2320 /// their declaration — the shape a second child model would produce if it
2321 /// declared the same two parents the other way round.
2322 fn two_legs() -> Vec<CounterCacheSpec<Dummy>> {
2323 let mut users = spec(false);
2324 users.parent_table = "users";
2325 users.counter_column = "sent_count";
2326 let mut posts = spec(false);
2327 posts.parent_table = "posts";
2328 posts.counter_column = "comment_count";
2329 // Declared users-first; `posts` sorts first.
2330 vec![users, posts]
2331 }
2332
2333 #[test]
2334 fn deltas_go_out_in_a_globally_stable_lock_order() {
2335 // The order has to come from the schema, not from either model's
2336 // declaration order: two child models declaring the same two parents in
2337 // opposite orders would otherwise take the two row locks in opposite
2338 // orders and deadlock, and the generated transactions do not retry.
2339 let specs = two_legs();
2340 let ordered = fold_and_order(
2341 &specs,
2342 vec![
2343 (0, 5, 1, TenantScope::SameTenantAsChild(100)),
2344 (1, 9, 1, TenantScope::SameTenantAsChild(100)),
2345 (1, 2, 1, TenantScope::SameTenantAsChild(100)),
2346 ],
2347 );
2348 let keys: Vec<(&str, i64)> = ordered
2349 .iter()
2350 .map(|(i, parent_id, _, _)| (specs[*i].parent_table, *parent_id))
2351 .collect();
2352 assert_eq!(keys, vec![("posts", 2), ("posts", 9), ("users", 5)]);
2353
2354 // …and the same for the paths that resolve the parent inside the
2355 // statement, which can only order by table.
2356 assert_eq!(specs_in_lock_order(&specs), vec![1, 0]);
2357 }
2358
2359 #[test]
2360 fn contributions_to_one_parent_fold_into_a_single_statement() {
2361 let specs = two_legs();
2362 // Re-parenting away from and back to the same parent nets out; the two
2363 // remaining moves stay one statement each.
2364 let ordered = fold_and_order(
2365 &specs,
2366 vec![
2367 (1, 2, -1, TenantScope::SameTenantAsChild(10)),
2368 (1, 2, 1, TenantScope::SameTenantAsChild(11)),
2369 (1, 7, 1, TenantScope::SameTenantAsChild(12)),
2370 (1, 7, 1, TenantScope::SameTenantAsChild(13)),
2371 (0, 4, -1, TenantScope::SameTenantAsChild(14)),
2372 ],
2373 );
2374 assert_eq!(
2375 ordered,
2376 vec![
2377 (1, 7, 2, TenantScope::SameTenantAsChild(12)),
2378 (0, 4, -1, TenantScope::SameTenantAsChild(14))
2379 ]
2380 );
2381 }
2382
2383 #[test]
2384 fn a_net_total_no_single_delta_can_carry_goes_out_in_same_sign_pieces() {
2385 // Two children each weighing `i64::MAX` onto one parent: the sum is
2386 // not an `i64`, so a wrapping fold would apply `-2`. The net total is
2387 // carried by two deltas of the same sign instead.
2388 let specs = two_legs();
2389 let ordered = fold_and_order(
2390 &specs,
2391 vec![
2392 (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2393 (1, 2, i64::MAX, TenantScope::SameTenantAsChild(11)),
2394 ],
2395 );
2396 assert_eq!(
2397 ordered,
2398 vec![
2399 (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2400 (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10))
2401 ]
2402 );
2403 // …and a total one past `i64::MAX` is `MAX` then `1`, never a wrapped
2404 // negative: the sum of every delta issued is the sum requested.
2405 let ordered = fold_and_order(
2406 &specs,
2407 vec![
2408 (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2409 (1, 2, -5, TenantScope::SameTenantAsChild(11)),
2410 (1, 2, 5, TenantScope::SameTenantAsChild(12)),
2411 (1, 2, 1, TenantScope::SameTenantAsChild(13)),
2412 ],
2413 );
2414 assert_eq!(
2415 ordered,
2416 vec![
2417 (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2418 (1, 2, 1, TenantScope::SameTenantAsChild(10))
2419 ]
2420 );
2421 }
2422
2423 #[test]
2424 fn weights_that_cancel_across_an_overflowing_intermediate_issue_nothing() {
2425 // `[MAX, MAX, -MAX, -MAX]` nets to zero. Segmenting at the overflow
2426 // point would have issued `+MAX` then `-MAX`, and a parent sitting at
2427 // `1` would have aborted the whole mutation on the first statement
2428 // even though its final value is exactly where it started.
2429 let specs = two_legs();
2430 let ordered = fold_and_order(
2431 &specs,
2432 vec![
2433 (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2434 (1, 2, i64::MAX, TenantScope::SameTenantAsChild(11)),
2435 (1, 2, -i64::MAX, TenantScope::SameTenantAsChild(12)),
2436 (1, 2, -i64::MAX, TenantScope::SameTenantAsChild(13)),
2437 ],
2438 );
2439 assert!(ordered.is_empty(), "{ordered:?}");
2440 // The same shape netting to a representable value is one delta.
2441 let ordered = fold_and_order(
2442 &specs,
2443 vec![
2444 (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2445 (1, 2, i64::MAX, TenantScope::SameTenantAsChild(11)),
2446 (1, 2, -i64::MAX, TenantScope::SameTenantAsChild(12)),
2447 (1, 2, -7, TenantScope::SameTenantAsChild(13)),
2448 ],
2449 );
2450 assert_eq!(
2451 ordered,
2452 vec![(1, 2, i64::MAX - 7, TenantScope::SameTenantAsChild(10))]
2453 );
2454 }
2455
2456 #[test]
2457 fn a_weight_edit_whose_difference_does_not_fit_goes_out_as_two_deltas() {
2458 // `i64::MIN` edited to `i64::MAX` on the same parent: the mathematical
2459 // difference is 2^64 - 1, which is no `i64`, so the old weight comes
2460 // off and the new one goes on as separate statements. Folding keeps
2461 // them apart for the same reason.
2462 let mut out = Vec::new();
2463 push_diff(
2464 0,
2465 Some(captured(7, i64::MIN)),
2466 Some(captured(7, i64::MAX)),
2467 1,
2468 &mut out,
2469 );
2470 assert_eq!(
2471 out,
2472 vec![
2473 (0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
2474 (0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
2475 (0, 7, i64::MAX, TenantScope::SameTenantAsChild(1))
2476 ]
2477 );
2478 // Folded, the three pieces net to 2^64 - 1 and go out as same-sign
2479 // deltas: the parent climbs monotonically to its final value.
2480 let specs = two_legs();
2481 assert_eq!(
2482 fold_and_order(&specs, out.clone()),
2483 vec![
2484 (0, 7, i64::MAX, TenantScope::SameTenantAsChild(1)),
2485 (0, 7, i64::MAX, TenantScope::SameTenantAsChild(1)),
2486 (0, 7, 1, TenantScope::SameTenantAsChild(1))
2487 ]
2488 );
2489
2490 // An ordinary edit is still one delta for the difference.
2491 let mut out = Vec::new();
2492 push_diff(0, Some(captured(7, 3)), Some(captured(7, 10)), 1, &mut out);
2493 assert_eq!(out, vec![(0, 7, 7, TenantScope::SameTenantAsChild(1))]);
2494 }
2495
2496 #[test]
2497 fn removing_the_one_unnegatable_weight_splits_it_in_two() {
2498 // `-i64::MIN` does not exist. Deleting (or re-parenting) a child that
2499 // weighs exactly that much takes it off the old parent as two halves
2500 // whose sum is the whole, rather than wrapping to `i64::MIN` again and
2501 // moving the parent the wrong way by 2^63.
2502 let mut out = Vec::new();
2503 push_diff(0, Some(captured(7, i64::MIN)), None, 1, &mut out);
2504 assert_eq!(
2505 out,
2506 vec![
2507 (0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
2508 (0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1))
2509 ]
2510 );
2511 assert_eq!(
2512 out.iter()
2513 .map(|(_, _, delta, _)| i128::from(*delta))
2514 .sum::<i128>(),
2515 -i128::from(i64::MIN)
2516 );
2517
2518 let mut out = Vec::new();
2519 push_diff(
2520 0,
2521 Some(captured(7, i64::MIN)),
2522 Some(captured(8, 4)),
2523 1,
2524 &mut out,
2525 );
2526 assert_eq!(
2527 out,
2528 vec![
2529 (0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
2530 (0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
2531 (0, 8, 4, TenantScope::SameTenantAsChild(1))
2532 ]
2533 );
2534 }
2535
2536 #[test]
2537 fn a_tenant_scoped_run_is_ordered_so_no_prefix_leaves_the_span_of_zero_and_total() {
2538 // The leg cannot fold (each delta is guarded by its own child's
2539 // tenant), so the order is what keeps `[MAX, MAX, -MAX]` from passing
2540 // through `2 * MAX` on the way to its representable total.
2541 let mut specs = two_legs();
2542 specs[1].tenant_column = Some("tenant_id");
2543 let ordered = fold_and_order(
2544 &specs,
2545 vec![
2546 (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2547 (1, 2, i64::MAX, TenantScope::SameTenantAsChild(11)),
2548 (1, 2, -i64::MAX, TenantScope::SameTenantAsChild(12)),
2549 ],
2550 );
2551 let deltas: Vec<i64> = ordered.iter().map(|(_, _, d, _)| *d).collect();
2552 assert_eq!(deltas, vec![i64::MAX, -i64::MAX, i64::MAX]);
2553 // …and symmetrically for a negative total.
2554 let ordered = fold_and_order(
2555 &specs,
2556 vec![
2557 (1, 2, -i64::MAX, TenantScope::SameTenantAsChild(10)),
2558 (1, 2, -i64::MAX, TenantScope::SameTenantAsChild(11)),
2559 (1, 2, i64::MAX, TenantScope::SameTenantAsChild(12)),
2560 ],
2561 );
2562 let deltas: Vec<i64> = ordered.iter().map(|(_, _, d, _)| *d).collect();
2563 assert_eq!(deltas, vec![-i64::MAX, i64::MAX, -i64::MAX]);
2564 // Every child keeps its own witness, and other parents are untouched.
2565 let witnesses: Vec<TenantScope> = ordered.iter().map(|(_, _, _, w)| w.clone()).collect();
2566 assert_eq!(
2567 witnesses,
2568 vec![
2569 TenantScope::SameTenantAsChild(10),
2570 TenantScope::SameTenantAsChild(12),
2571 TenantScope::SameTenantAsChild(11)
2572 ]
2573 );
2574 }
2575
2576 fn captured_in(parent: i64, contrib: i64, tenant: Option<&str>) -> Captured {
2577 Captured {
2578 parent,
2579 contrib,
2580 tenant: tenant.map(str::to_owned),
2581 }
2582 }
2583
2584 #[test]
2585 fn a_tenant_change_takes_the_old_weight_off_under_the_old_tenant() {
2586 // The row now says the new tenant, so a live predicate would look for
2587 // the old parent in the wrong tenant and miss it; the removal is
2588 // scoped by the tenant the capture read. The addition is live.
2589 let live = TenantScope::SameTenantAsChild(1);
2590 let was_a = TenantScope::Captured(Some("a".to_owned()));
2591 let mut out = Vec::new();
2592 push_diff(
2593 0,
2594 Some(captured_in(7, 3, Some("a"))),
2595 Some(captured_in(7, 3, Some("b"))),
2596 1,
2597 &mut out,
2598 );
2599 assert_eq!(
2600 out,
2601 vec![(0, 7, -3, was_a.clone()), (0, 7, 3, live.clone())]
2602 );
2603
2604 // Re-parented and moved at once: the same two statements.
2605 let mut out = Vec::new();
2606 push_diff(
2607 0,
2608 Some(captured_in(7, 3, Some("a"))),
2609 Some(captured_in(8, 3, Some("b"))),
2610 1,
2611 &mut out,
2612 );
2613 assert_eq!(
2614 out,
2615 vec![(0, 7, -3, was_a.clone()), (0, 8, 3, live.clone())]
2616 );
2617
2618 // A NULL old tenant is a captured NULL, not "unknown".
2619 let mut out = Vec::new();
2620 push_diff(
2621 0,
2622 Some(captured_in(7, 1, None)),
2623 Some(captured_in(7, 1, Some("b"))),
2624 1,
2625 &mut out,
2626 );
2627 assert_eq!(
2628 out,
2629 vec![
2630 (0, 7, -1, TenantScope::Captured(None)),
2631 (0, 7, 1, live.clone())
2632 ]
2633 );
2634
2635 // Same tenant: the diff is what it always was, one live delta.
2636 let mut out = Vec::new();
2637 push_diff(
2638 0,
2639 Some(captured_in(7, 3, Some("a"))),
2640 Some(captured_in(7, 10, Some("a"))),
2641 1,
2642 &mut out,
2643 );
2644 assert_eq!(out, vec![(0, 7, 7, live)]);
2645
2646 // Unparented after a tenant change: the removal is still under the old
2647 // tenant, and nothing is added.
2648 let mut out = Vec::new();
2649 push_diff(0, Some(captured_in(7, 3, Some("a"))), None, 1, &mut out);
2650 assert_eq!(out, vec![(0, 7, -3, was_a)]);
2651 }
2652
2653 #[test]
2654 fn only_a_table_with_a_leg_onto_itself_is_serialized_and_only_once() {
2655 // Two ordinary legs: nothing to take.
2656 assert!(self_referential_tables(&two_legs()).is_empty());
2657 // Two legs onto their own table, plus one ordinary: that table once,
2658 // and tables in a stable order across mutations.
2659 let mut replies = spec(false);
2660 replies.parent_table = "comments";
2661 replies.counter_column = "reply_count";
2662 let mut likes = spec(false);
2663 likes.parent_table = "comments";
2664 likes.counter_column = "like_count";
2665 let mut users = spec(false);
2666 users.child_table = "avatars";
2667 users.parent_table = "avatars";
2668 assert_eq!(
2669 self_referential_tables(&[users, replies, spec(false), likes]),
2670 vec!["avatars", "comments"]
2671 );
2672 }
2673
2674 #[test]
2675 fn a_captured_tenant_scopes_the_parent_by_its_text() {
2676 let mut tenanted = spec(false);
2677 tenanted.tenant_column = Some("tenant_id");
2678 assert_eq!(
2679 tenant_predicate_captured(&view(&tenanted)),
2680 format!(" AND CAST(\"posts\".\"tenant_id\" AS TEXT) {IS_NOT_DISTINCT_FROM} {PH3}")
2681 );
2682 // Without a tenant column there is no predicate and no bind.
2683 assert_eq!(tenant_predicate_captured(&view(&spec(false))), "");
2684 }
2685
2686 #[test]
2687 fn a_tenant_scoped_leg_keeps_one_statement_per_child() {
2688 // Folding behind a single arbitrary witness would either sweep
2689 // cross-tenant children into the delta or drop legitimate ones, so a
2690 // tenant-scoped spec stays unfolded — still in the global order.
2691 let mut specs = two_legs();
2692 specs[1].tenant_column = Some("tenant_id");
2693 let ordered = fold_and_order(
2694 &specs,
2695 vec![
2696 (1, 3, 1, TenantScope::SameTenantAsChild(20)),
2697 (1, 3, 1, TenantScope::SameTenantAsChild(21)),
2698 ],
2699 );
2700 assert_eq!(
2701 ordered,
2702 vec![
2703 (1, 3, 1, TenantScope::SameTenantAsChild(20)),
2704 (1, 3, 1, TenantScope::SameTenantAsChild(21))
2705 ]
2706 );
2707 }
2708
2709 #[test]
2710 fn a_tenant_discriminator_is_matched_null_safely() {
2711 // A nullable discriminator with NULL on both sides is the same tenant
2712 // (namely none), but `=` yields NULL there, so the maintenance would
2713 // silently do nothing — a counter that quietly stops moving is worse
2714 // than one that errors.
2715 let mut tenanted = spec(false);
2716 tenanted.tenant_column = Some("tenant_id");
2717
2718 let joined = tenant_predicate_joined(&view(&tenanted));
2719 assert!(joined.contains(IS_NOT_DISTINCT_FROM), "{joined}");
2720 assert!(
2721 !joined.contains("\"tenant_id\" = "),
2722 "plain equality is not NULL-safe: {joined}"
2723 );
2724
2725 // The parent-keyed form still requires the child row to exist, so a
2726 // missing child remains a no-op rather than matching every untenanted
2727 // parent the way a scalar sub-select would.
2728 let keyed = tenant_predicate(&view(&tenanted), 7);
2729 assert!(keyed.contains("EXISTS"), "{keyed}");
2730 assert!(keyed.contains(IS_NOT_DISTINCT_FROM), "{keyed}");
2731 assert!(keyed.contains("\"id\" = 7"), "{keyed}");
2732
2733 // Still nothing at all for an association that declares no tenant.
2734 assert_eq!(tenant_predicate_joined(&view(&spec(false))), "");
2735 assert_eq!(tenant_predicate(&view(&spec(false)), 7), "");
2736 }
2737
2738 #[test]
2739 fn a_counter_column_named_after_a_sql_keyword_still_produces_valid_sql() {
2740 // `counter_cache = "order"` is a legal identifier and a legal column
2741 // name, but `SET order = order + $1` is a syntax error on both backends.
2742 // Quoting every interpolated identifier is what keeps the choice of
2743 // column name from being able to break the generated statements.
2744 let mut keyword = spec(false);
2745 keyword.counter_column = "order";
2746 keyword.parent_table = "group";
2747 let sql = recompute_update_sql(&view(&keyword), "1");
2748 assert!(
2749 sql.starts_with("UPDATE \"group\" SET \"order\" = "),
2750 "{sql}"
2751 );
2752 assert!(sql.contains("\"group\".\"order\" "), "{sql}");
2753 assert!(
2754 !sql.contains(" order "),
2755 "no bare keyword may survive: {sql}"
2756 );
2757 }
2758
2759 #[test]
2760 fn a_plain_counter_cache_keeps_its_pre_derivation_sql() {
2761 // The whole design rests on this: derivations reuse the counter-cache
2762 // paths, so a counter cache's own SQL must be untouched by them.
2763 let plain = view(&spec(false));
2764 assert!(plain.is_plain());
2765 assert!(plain.counts_rows());
2766 assert_eq!(filter_predicate(&plain, CHILD_ALIAS), "");
2767 assert_eq!(aggregate_expr(&plain, CHILD_ALIAS), "COUNT(*)");
2768 assert!(
2769 !recompute_update_sql(&plain, "1").contains("COALESCE"),
2770 "a plain counter cache must not grow a COALESCE"
2771 );
2772 assert_eq!(
2773 accumulator(&plain, "\"comment_count\""),
2774 "\"comment_count\"",
2775 "a plain counter cache reads its column bare"
2776 );
2777 assert!(
2778 !bulk_decrement_sql(&plain, "7").contains("COALESCE"),
2779 "a plain bulk decrement must not grow a COALESCE"
2780 );
2781 }
2782
2783 #[test]
2784 fn a_derivation_delta_never_writes_null_into_the_maintained_column() {
2785 // A derivation is adopted on an existing table, so the column can still
2786 // be NULL. `NULL +/- expr` is NULL, and a NULL maintained value cannot
2787 // be told apart from drift, so every adjusting statement coalesces.
2788 let mut filtered = spec(false);
2789 filtered.filter_sql = " AND ({c}.\"published\" = TRUE)";
2790 let filtered = view(&filtered);
2791 assert_eq!(
2792 accumulator(&filtered, "\"comment_count\""),
2793 "COALESCE(\"comment_count\", 0)"
2794 );
2795 assert!(
2796 bulk_decrement_sql(&filtered, "7")
2797 .contains("SET \"comment_count\" = COALESCE(\"comment_count\", 0) -"),
2798 "{}",
2799 bulk_decrement_sql(&filtered, "7")
2800 );
2801 let sum = view(&sum_spec());
2802 assert!(
2803 weighted_delta_by_child_id_sql(&sum, 1, false, "", "")
2804 .contains("SET \"visible_score\" = COALESCE(\"visible_score\", 0) +"),
2805 "the by-id weighted delta coalesces too"
2806 );
2807 }
2808
2809 #[test]
2810 fn a_lowered_contribution_is_either_one_or_a_single_child_column() {
2811 // The run-time backstop for `contrib_sql`: a structural check, so a
2812 // legal quoted literal is never mistaken for smuggled SQL.
2813 assert!(is_lowered_contribution("1"));
2814 assert!(is_lowered_contribution("{c}.\"score\""));
2815 assert!(!is_lowered_contribution("{c}.\"score\" + 1"));
2816 assert!(!is_lowered_contribution("(SELECT 1)"));
2817 assert!(!is_lowered_contribution(""));
2818 assert!(!is_lowered_contribution("{c}.\"a\"; DROP TABLE posts --\""));
2819 }
2820
2821 #[test]
2822 fn a_filter_may_carry_a_comment_marker_inside_a_quoted_literal() {
2823 // The previous substring scan panicked on this in debug builds: `--`
2824 // and `;` are inert inside a quoted literal, and the macro escapes the
2825 // literal it lowers, so only the identifiers and the contribution are
2826 // checked at run time.
2827 let mut filtered = spec(false);
2828 filtered.filter_sql = " AND ({c}.\"status\" = 'a -- b; c')";
2829 let filtered = view(&filtered);
2830 assert_spec_idents(&filtered);
2831 assert!(
2832 bulk_decrement_sql(&filtered, "7").contains("'a -- b; c'"),
2833 "the literal reaches the statement unaltered"
2834 );
2835 }
2836
2837 #[test]
2838 fn a_filtered_count_carries_its_filter_into_every_statement() {
2839 let mut filtered = spec(false);
2840 filtered.filter_sql = " AND ({c}.\"published\" = TRUE)";
2841 let filtered = view(&filtered);
2842 assert!(
2843 !filtered.is_plain(),
2844 "a filter is not a plain counter cache"
2845 );
2846 assert!(filtered.counts_rows(), "a filtered count still counts rows");
2847
2848 // The deltas: the filter sits in both sub-selects, so a rejected row
2849 // matches no parent and the statement writes nothing.
2850 let delete = bulk_decrement_sql(&filtered, "7,8");
2851 assert_eq!(
2852 delete,
2853 "UPDATE \"posts\" SET \"comment_count\" = \
2854 COALESCE(\"comment_count\", 0) - \
2855 (SELECT COUNT(*) FROM \"comments\" AS __autumn_cc_child \
2856 WHERE __autumn_cc_child.\"post_id\" = \"posts\".\"id\" \
2857 AND __autumn_cc_child.\"id\" IN (7,8) \
2858 AND (__autumn_cc_child.\"published\" = TRUE)) \
2859 WHERE \"posts\".\"id\" IN \
2860 (SELECT __autumn_cc_child.\"post_id\" FROM \"comments\" AS __autumn_cc_child \
2861 WHERE __autumn_cc_child.\"id\" IN (7,8) \
2862 AND __autumn_cc_child.\"post_id\" IS NOT NULL \
2863 AND (__autumn_cc_child.\"published\" = TRUE) \
2864 AND __autumn_cc_child.\"post_id\" = \"posts\".\"id\")"
2865 );
2866
2867 // The capture reads the contribution as a column, so a filter flip on an
2868 // unchanged parent is visible as `0 -> 1`.
2869 assert_eq!(
2870 contrib_case_expr(&filtered, CHILD_ALIAS),
2871 "CAST(CASE WHEN 1 = 1 AND (__autumn_cc_child.\"published\" = TRUE) \
2872 THEN 1 ELSE 0 END AS BIGINT)"
2873 );
2874 }
2875
2876 #[test]
2877 fn a_sum_recompute_assigns_the_summed_contribution() {
2878 let sum = view(&sum_spec());
2879 assert!(!sum.counts_rows());
2880 assert_eq!(
2881 recompute_update_sql(&sum, "1,2"),
2882 "UPDATE \"posts\" SET \"visible_score\" = \
2883 (SELECT COALESCE(SUM(__autumn_cc_child.\"score\"), 0) \
2884 FROM \"comments\" AS __autumn_cc_child \
2885 WHERE __autumn_cc_child.\"post_id\" = \"posts\".\"id\" \
2886 AND (__autumn_cc_child.\"published\" = TRUE)) \
2887 WHERE \"posts\".\"id\" IN (1,2) \
2888 AND \"posts\".\"visible_score\" "
2889 .to_owned()
2890 + IS_DISTINCT_FROM
2891 + " (SELECT COALESCE(SUM(__autumn_cc_child.\"score\"), 0) \
2892 FROM \"comments\" AS __autumn_cc_child \
2893 WHERE __autumn_cc_child.\"post_id\" = \"posts\".\"id\" \
2894 AND (__autumn_cc_child.\"published\" = TRUE))"
2895 );
2896 }
2897
2898 #[test]
2899 fn a_weighted_by_id_delta_reads_the_weight_from_the_row() {
2900 // The magnitude cannot be bound by the caller: only the database can see
2901 // the row the weight comes from. `COALESCE` keeps a NULL out of the
2902 // maintained column even so.
2903 let sql = weighted_delta_by_child_id_sql(&view(&sum_spec()), 9, false, "", "");
2904 assert!(
2905 sql.starts_with(
2906 "UPDATE \"posts\" SET \"visible_score\" = \
2907 COALESCE(\"visible_score\", 0) + \
2908 COALESCE((SELECT __autumn_cc_child.\"score\""
2909 ),
2910 "{sql}"
2911 );
2912 assert_eq!(
2913 sql.matches("__autumn_cc_child.\"id\" = 9").count(),
2914 2,
2915 "{sql}"
2916 );
2917 assert!(
2918 sql.contains(&format!(
2919 "IS NOT NULL AND (__autumn_cc_child.\"published\" = TRUE){FOR_UPDATE})"
2920 )),
2921 "{sql}"
2922 );
2923
2924 let down = weighted_delta_by_child_id_sql(&view(&sum_spec()), 9, true, "", "");
2925 assert!(
2926 down.contains("\"visible_score\" = COALESCE(\"visible_score\", 0) - COALESCE("),
2927 "{down}"
2928 );
2929 }
2930
2931 #[test]
2932 fn a_filter_flip_on_the_same_parent_is_one_delta() {
2933 let mut out = Vec::new();
2934 // Unpublished (0) -> published (1) with the parent unchanged.
2935 push_diff(0, Some(captured(4, 0)), Some(captured(4, 1)), 11, &mut out);
2936 assert_eq!(out, vec![(0, 4, 1, TenantScope::SameTenantAsChild(11))]);
2937
2938 // A rejected row that stays rejected moves nothing, even across a
2939 // reparent: neither side has a weight to move.
2940 out.clear();
2941 push_diff(0, Some(captured(4, 0)), Some(captured(5, 0)), 11, &mut out);
2942 assert!(out.is_empty(), "{out:?}");
2943
2944 // A reparent of a qualifying row moves both ends.
2945 out.clear();
2946 push_diff(0, Some(captured(4, 3)), Some(captured(5, 3)), 11, &mut out);
2947 assert_eq!(
2948 out,
2949 vec![
2950 (0, 4, -3, TenantScope::SameTenantAsChild(11)),
2951 (0, 5, 3, TenantScope::SameTenantAsChild(11))
2952 ]
2953 );
2954
2955 // A weight edit on an unchanged parent is the difference only.
2956 out.clear();
2957 push_diff(0, Some(captured(4, 3)), Some(captured(4, 10)), 11, &mut out);
2958 assert_eq!(out, vec![(0, 4, 7, TenantScope::SameTenantAsChild(11))]);
2959 }
2960
2961 #[test]
2962 fn drift_is_one_aggregate_over_the_parent_table() {
2963 let sql = drift_sql(&view(&sum_spec()), 10_000);
2964 assert!(
2965 sql.starts_with("SELECT COUNT(*) AS count FROM (SELECT 1 AS drifted FROM "),
2966 "{sql}"
2967 );
2968 assert!(sql.contains(IS_DISTINCT_FROM), "{sql}");
2969 // The cap is on the parents examined, not on the matches: a healthy
2970 // table has no match, so an output-side LIMIT would still evaluate the
2971 // correlated aggregate for every parent before giving up.
2972 assert!(
2973 sql.contains(
2974 "FROM (SELECT * FROM \"posts\" ORDER BY \"posts\".\"id\" DESC LIMIT 10000) AS \"posts\" WHERE"
2975 ),
2976 "the window is materialised under the table's own name: {sql}"
2977 );
2978 assert!(sql.ends_with(") AS __autumn_cc_drift"), "{sql}");
2979 }
2980
2981 #[test]
2982 fn a_model_without_an_inherent_shadow_resolves_to_the_empty_blanket() {
2983 // A `const` block, because the whole point of the flag is that it folds
2984 // at compile time — `assert!` on a constant is a clippy error.
2985 const { assert!(!Dummy::HAS_COUNTER_CACHES) };
2986 assert!(Dummy::counter_caches().is_empty());
2987 }
2988}