Skip to main content

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. [`sum_expr`] for a weighted sum.
451fn aggregate_expr(view: &SqlView, alias: &str) -> String {
452    if view.counts_rows() {
453        "COUNT(*)".to_owned()
454    } else {
455        sum_expr(&contrib_expr(view, alias))
456    }
457}
458
459/// Sums `contrib`. `SUM` of no rows is NULL, so `COALESCE` gives 0. Postgres
460/// sums `bigint` as `numeric`, so a partial sum cannot overflow.
461#[cfg(not(feature = "sqlite"))]
462pub fn sum_expr(contrib: &str) -> String {
463    format!("COALESCE(SUM({contrib}), 0)")
464}
465
466/// An exact `SQLite` sum that no row order can overflow (#2663).
467///
468/// `SQLite`'s `SUM` raises `integer overflow` when a partial sum is outside
469/// `i64`, even when the total is in range. The delta paths add in `i128`. The
470/// set-based paths must accept the same rows. See [`split_join`].
471///
472/// See also `money::ledger::split_sum`, which joins its halves in Rust.
473#[cfg(feature = "sqlite")]
474pub fn sum_expr(contrib: &str) -> String {
475    let (hi, lo) = split_sums(contrib);
476    split_join(&hi, &lo)
477}
478
479/// The low 32 bits of a split `i64`.
480#[cfg(feature = "sqlite")]
481const LOW_MASK: u32 = u32::MAX;
482
483/// `(hi, lo)` sums of `contrib`, with the total `hi * 2^32 + lo`.
484///
485/// Each row splits as `x = (x >> 32) * 2^32 + (x & 0xFFFFFFFF)`. Each half-sum
486/// stays in `i64` for fewer than 2^31 rows.
487#[cfg(feature = "sqlite")]
488fn split_sums(contrib: &str) -> (String, String) {
489    (
490        format!("COALESCE(SUM(({contrib}) >> 32), 0)"),
491        format!("COALESCE(SUM(({contrib}) & {LOW_MASK}), 0)"),
492    )
493}
494
495/// `hi * 2^32 + lo` as one `i64`, or `integer overflow` when it is out of range.
496///
497/// The carry of `lo` goes into `hi`, and the low part that stays,
498/// `lo & 0xFFFFFFFF`, is in `[0, 2^32)`. Thus the total is in `i64` only when
499/// the carried `hi` is in `i32`. The `CASE` checks this. Without the check,
500/// `SQLite` would change the product to a REAL.
501#[cfg(feature = "sqlite")]
502fn split_join(hi: &str, lo: &str) -> String {
503    let (min, max, radix) = (i32::MIN, i32::MAX, 1_i64 << 32);
504    let hi = format!("({hi} + (({lo}) >> 32))");
505    // `abs(i64::MIN)` raises `integer overflow`, as `SUM` did. It is spelled
506    // `-MAX - 1`: SQLite reads `9223372036854775808` as a REAL, and `abs` of a
507    // REAL does not raise.
508    format!(
509        "CASE WHEN {hi} BETWEEN {min} AND {max} \
510         THEN {hi} * {radix} + (({lo}) & {LOW_MASK}) \
511         ELSE abs(-9223372036854775807 - 1) END"
512    )
513}
514
515/// `CASE WHEN <filter> THEN <contrib> ELSE 0 END`, for the statements that read
516/// a row's contribution as a column.
517///
518/// The `1 = 1` seed absorbs `filter_sql`'s leading ` AND `, which every other
519/// caller concatenates onto a predicate it already has.
520///
521/// The `CAST` is not cosmetic: this value is *decoded* into an `i64`, and a bare
522/// integer literal is `integer` on Postgres, so the driver would refuse the
523/// four bytes it got where it expected eight. Casting also normalises a legacy
524/// 32-bit summed column to the width the maintained value is read at.
525fn contrib_case_expr(view: &SqlView, alias: &str) -> String {
526    let contrib = contrib_expr(view, alias);
527    if view.filter_sql.is_empty() {
528        return format!("CAST({contrib} AS BIGINT)");
529    }
530    let filter = filter_predicate(view, alias);
531    format!("CAST(CASE WHEN 1 = 1{filter} THEN {contrib} ELSE 0 END AS BIGINT)")
532}
533
534/// `AND <parent>.<tenant> = __autumn_cc_child.<tenant>`, for the statements that
535/// already have the child row joined in a sub-select. Empty when the association
536/// declares no `counter_cache_tenant`.
537///
538/// The predicate must be applied to **every** delta path or it makes things
539/// worse rather than better: scoping only the increment would leave the matching
540/// decrement unscoped, so a cross-tenant foreign key would drive another
541/// tenant's counter down without ever driving it up.
542///
543/// `AND EXISTS (SELECT 1 FROM <child> x WHERE x.<pk> = N AND <parent>.tenant IS
544/// NOT DISTINCT FROM x.tenant)` for the parent-keyed statements, which have no
545/// child row to join, empty otherwise. The `EXISTS` (rather than a scalar
546/// sub-select) keeps the statement a no-op when the child row is absent, which
547/// is what a missing child has always meant here.
548///
549/// Deliberately a sub-select on the child row rather than a bound ambient
550/// tenant: the invariant being enforced is "the parent is in the same tenant as
551/// the child that names it", which is exactly what makes a cross-tenant foreign
552/// key move nothing. It needs no tenant plumbing through the generated code, and
553/// it is correct for `across_tenants()` callers too.
554fn tenant_predicate_joined(view: &SqlView) -> String {
555    let Some(tenant_column) = view.tenant_column else {
556        return String::new();
557    };
558    let tenant_column = quote_ident(tenant_column);
559    let parent_table = quote_ident(view.parent_table);
560    format!(
561        " AND {parent_table}.{tenant_column} {IS_NOT_DISTINCT_FROM} \
562         {CHILD_ALIAS}.{tenant_column}"
563    )
564}
565
566/// `AND CAST(<parent>.<tenant> AS TEXT) IS NOT DISTINCT FROM $3`, for a delta
567/// scoped by a tenant the pre-update capture read (see
568/// [`TenantScope::Captured`]). Both sides are text so one bind serves every
569/// tenant column type; `IS NOT DISTINCT FROM` matches a NULL tenant to a NULL
570/// bind the way the joined predicate does.
571fn tenant_predicate_captured(view: &SqlView) -> String {
572    let Some(tenant_column) = view.tenant_column else {
573        return String::new();
574    };
575    let tenant_column = quote_ident(tenant_column);
576    let parent_table = quote_ident(view.parent_table);
577    format!(" AND CAST({parent_table}.{tenant_column} AS TEXT) {IS_NOT_DISTINCT_FROM} {PH3}")
578}
579
580fn tenant_predicate(view: &SqlView, child_id: i64) -> String {
581    let Some(tenant_column) = view.tenant_column else {
582        return String::new();
583    };
584    let tenant_column = quote_ident(tenant_column);
585    let Quoted {
586        child_table,
587        child_pk,
588        parent_table,
589        ..
590    } = quoted(view);
591    format!(
592        " AND EXISTS \
593         (SELECT 1 FROM {child_table} AS {CHILD_ALIAS}_t \
594          WHERE {CHILD_ALIAS}_t.{child_pk} = {child_id} \
595            AND {parent_table}.{tenant_column} {IS_NOT_DISTINCT_FROM} \
596                {CHILD_ALIAS}_t.{tenant_column})"
597    )
598}
599
600/// The left-hand side of a `SET c = <acc> +/- ...` adjustment.
601///
602/// A plain counter cache reads the column bare, so its SQL stays byte-identical
603/// to what it was before derivations existed. A derivation wraps it in
604/// `COALESCE(c, 0)`: a derivation is adopted on an existing table, where the
605/// column can still be NULL, and `NULL +/- expr` is NULL. One such adjustment
606/// would poison the column, and a NULL maintained value is the one outcome no
607/// repair can tell apart from legitimate drift.
608fn accumulator(view: &SqlView, counter_column: &str) -> String {
609    if view.is_plain() {
610        counter_column.to_owned()
611    } else {
612        format!("COALESCE({counter_column}, 0)")
613    }
614}
615
616/// `AND __autumn_cc_child.deleted_at IS NULL` (or `IS NOT NULL`) for a
617/// soft-deleting child, empty for a child with no `deleted_at` column.
618fn live_predicate(view: &SqlView, want_live: bool) -> String {
619    if !view.child_soft_delete {
620        return String::new();
621    }
622    let op = if want_live { "IS NULL" } else { "IS NOT NULL" };
623    let deleted_at = quote_ident(DELETED_AT);
624    format!(" AND {CHILD_ALIAS}.{deleted_at} {op}")
625}
626
627/// Take, for the rest of the transaction, the lock that serializes every
628/// counter-cached mutation on a table with a leg onto itself.
629///
630/// Call it before a raw write of your own on such a table, ahead of the
631/// `INSERT`, `UPDATE` or `DELETE` itself, and before
632/// [`counter_cache_after_insert_by_id`] or its siblings; see the note on that
633/// function.
634///
635/// Onto its own table, a mutation locks a child row and then a parent row of
636/// the same table. Two such mutations can want each other's rows: one moves
637/// node A under B (holds A, waits for B) while another moves B under A (holds
638/// B, waits for A), and the database breaks the cycle by aborting one, which
639/// the generated repository transaction does not retry. Ordering the parent
640/// deltas cannot help, since each child lock is already held. So every such
641/// mutation first takes one transaction-scoped advisory lock keyed by the
642/// table, before any row lock, and two of them take turns instead of crossing.
643/// Every hook here takes it on entry, and every generated mutation that
644/// row-locks before its hook runs (`upsert_many`'s `FOR UPDATE` load, the
645/// hooked delete's load for `before_delete`, `delete_many`'s preload, the
646/// dependent cascade's child lock, a retention sweep's batch lock) calls this
647/// at the top of its transaction too, so the advisory lock always comes before
648/// the first row lock; the hook's later take is then a re-entrant no-op. A
649/// table without a self-referential leg costs nothing here. `SQLite`
650/// serializes writers with `BEGIN IMMEDIATE` already, so it has nothing to
651/// take.
652///
653/// # Errors
654///
655/// Propagates any database error from the lock statement.
656pub async fn counter_cache_serialize_self_referential<M: 'static>(
657    conn: &mut RuntimeConnection,
658    specs: &[CounterCacheSpec<M>],
659) -> AutumnResult<()> {
660    for table in self_referential_tables(specs) {
661        serialize_table(conn, table).await?;
662    }
663    Ok(())
664}
665
666/// The tables of `specs` with a leg onto themselves, each once, in a stable
667/// order (so two mutations over several such tables take the locks in the
668/// same order).
669fn self_referential_tables<M: 'static>(specs: &[CounterCacheSpec<M>]) -> Vec<&'static str> {
670    let mut tables: Vec<&'static str> = specs
671        .iter()
672        .filter(|spec| spec.parent_table == spec.child_table)
673        .map(|spec| spec.child_table)
674        .collect();
675    tables.sort_unstable();
676    tables.dedup();
677    tables
678}
679
680#[cfg(not(feature = "sqlite"))]
681async fn serialize_table(conn: &mut RuntimeConnection, table: &str) -> AutumnResult<()> {
682    // The two-key form keeps this out of the single-key namespace the
683    // framework's migration and job locks use; the first key names this
684    // purpose, the second the table.
685    diesel::sql_query(
686        "SELECT pg_advisory_xact_lock(hashtext('autumn_counter_cache'), hashtext($1))",
687    )
688    .bind::<Text, _>(table)
689    .execute(conn)
690    .await
691    .map_err(AutumnError::from)?;
692    Ok(())
693}
694
695#[cfg(feature = "sqlite")]
696#[allow(clippy::unused_async, reason = "one call shape for both backends")]
697async fn serialize_table(_conn: &mut RuntimeConnection, _table: &str) -> AutumnResult<()> {
698    Ok(())
699}
700
701/// Refuse counter upkeep on a ledgered parent. The `UPDATE` records no revision
702/// (#2319).
703fn refuse_ledgered_parent(view: &SqlView) -> AutumnResult<()> {
704    crate::ledger::refuse_out_of_band_write(view.parent_table, "counter cache")
705}
706
707/// Apply `delta` to one parent's counter with a single atomic statement.
708///
709/// This is the primitive AC5 rests on: `SET c = c + $1` is resolved by the
710/// database, so N concurrent callers commute and none can lose another's update
711/// the way a `SELECT` + `UPDATE` round trip would.
712///
713/// # Errors
714///
715/// Propagates any database error from the `UPDATE`.
716pub async fn counter_cache_apply_delta<M: 'static>(
717    conn: &mut RuntimeConnection,
718    spec: &CounterCacheSpec<M>,
719    parent_id: i64,
720    delta: i64,
721    scope: TenantScope,
722) -> AutumnResult<()> {
723    let view = view(spec);
724    assert_spec_idents(&view);
725    refuse_ledgered_parent(&view)?;
726    let Quoted {
727        parent_table,
728        parent_pk,
729        counter_column,
730        ..
731    } = quoted(&view);
732    let tenant = match &scope {
733        TenantScope::SameTenantAsChild(child_id) => tenant_predicate(&view, *child_id),
734        TenantScope::Captured(_) => tenant_predicate_captured(&view),
735        TenantScope::Unscoped => String::new(),
736    };
737    let accumulator = accumulator(&view, &counter_column);
738    let sql = format!(
739        "UPDATE {parent_table} SET {counter_column} = {accumulator} + {PH1} \
740         WHERE {parent_table}.{parent_pk} = {PH2}{tenant}"
741    );
742    let query = diesel::sql_query(sql)
743        .bind::<BigInt, _>(delta)
744        .bind::<BigInt, _>(parent_id);
745    // The captured tenant is a third bind, but only when the leg has a tenant
746    // column: without one the predicate is empty and so is the bind list.
747    if let (TenantScope::Captured(tenant), true) = (scope, view.tenant_column.is_some()) {
748        query
749            .bind::<Nullable<Text>, _>(tenant)
750            .execute(conn)
751            .await
752            .map_err(AutumnError::from)?;
753        return Ok(());
754    }
755    query.execute(conn).await.map_err(AutumnError::from)?;
756    Ok(())
757}
758
759/// Apply `delta` to the parent of the child row `child_id`, resolving the parent
760/// through a sub-select so no round trip is needed and a `NULL` foreign key is a
761/// no-op (`IN (NULL)` matches nothing).
762///
763/// `require_live` scopes the sub-select to a live child (`deleted_at IS NULL`)
764/// for a soft-deleting model, which is what makes a repeated soft delete — or a
765/// hard delete of an already-soft-deleted row — decrement exactly once.
766/// `require_soft_deleted` is its mirror, used by `restore`.
767///
768/// For a `#[derivation]` the magnitude cannot come from the caller: the weight
769/// is a property of the child row, which only the database can see here. So
770/// `delta` supplies the **sign** and the statement reads the contribution
771/// itself. The filter sits in both sub-selects, so a row the filter rejects
772/// matches no parent and the statement is a no-op.
773///
774/// # Errors
775///
776/// Propagates any database error from the `UPDATE`.
777#[doc(hidden)]
778pub async fn counter_cache_apply_delta_by_child_id<M: 'static>(
779    conn: &mut RuntimeConnection,
780    spec: &CounterCacheSpec<M>,
781    child_id: i64,
782    delta: i64,
783    child_state: ChildState,
784) -> AutumnResult<()> {
785    let view = view(spec);
786    assert_spec_idents(&view);
787    refuse_ledgered_parent(&view)?;
788    let state_predicate = match child_state {
789        ChildState::Any => String::new(),
790        ChildState::Live => live_predicate(&view, true),
791        ChildState::SoftDeleted => live_predicate(&view, false),
792    };
793    // The parent is resolved by a sub-select on the child row, so the tenant
794    // check is a correlated comparison in the outer `WHERE`. It names the child
795    // alias, which is only in scope for a sub-select the outer statement
796    // correlates with, so the whole predicate moves into a second sub-select.
797    let tenant = tenant_predicate(&view, child_id);
798
799    if !view.is_plain() {
800        debug_assert!(
801            delta == 1 || delta == -1,
802            "a derivation delta carries only a sign; the weight comes from the row"
803        );
804        let sql =
805            weighted_delta_by_child_id_sql(&view, child_id, delta < 0, &state_predicate, &tenant);
806        diesel::sql_query(sql)
807            .execute(conn)
808            .await
809            .map_err(AutumnError::from)?;
810        return Ok(());
811    }
812
813    let Quoted {
814        child_table,
815        child_pk,
816        fk_column,
817        parent_table,
818        parent_pk,
819        counter_column,
820        ..
821    } = quoted(&view);
822    let sql = format!(
823        "UPDATE {parent_table} SET {counter_column} = {counter_column} + {PH1} \
824         WHERE {parent_table}.{parent_pk} IN \
825           (SELECT {CHILD_ALIAS}.{fk_column} FROM {child_table} AS {CHILD_ALIAS} \
826            WHERE {CHILD_ALIAS}.{child_pk} = {PH2} \
827              AND {CHILD_ALIAS}.{fk_column} IS NOT NULL{state_predicate}{FOR_UPDATE}){tenant}"
828    );
829    diesel::sql_query(sql)
830        .bind::<BigInt, _>(delta)
831        .bind::<BigInt, _>(child_id)
832        .execute(conn)
833        .await
834        .map_err(AutumnError::from)?;
835    Ok(())
836}
837
838/// The `#[derivation]` form of [`counter_cache_apply_delta_by_child_id`]: the
839/// weight is read from the child row instead of bound by the caller.
840///
841/// `child_id` is inlined rather than bound, so the two occurrences stay one
842/// statement on both backends (`SQLite`'s `?` is positional, so a second bind
843/// would be required there and not on Postgres). It is an `i64`, so its decimal
844/// rendering cannot contain SQL syntax, the same type-level guarantee
845/// [`id_list`] rests on.
846///
847/// `COALESCE` is belt-and-braces: the outer `WHERE` already restricts the
848/// statement to parents whose child row qualifies, so the sub-select cannot be
849/// empty. But a NULL in the summed column would still poison the column, and a
850/// NULL maintained value is the one outcome no repair can distinguish from
851/// legitimate drift.
852fn weighted_delta_by_child_id_sql(
853    view: &SqlView,
854    child_id: i64,
855    subtract: bool,
856    state_predicate: &str,
857    tenant: &str,
858) -> String {
859    let Quoted {
860        child_table,
861        child_pk,
862        fk_column,
863        parent_table,
864        parent_pk,
865        counter_column,
866        ..
867    } = quoted(view);
868    let filter = filter_predicate(view, CHILD_ALIAS);
869    let contrib = contrib_expr(view, CHILD_ALIAS);
870    let sign = if subtract { "-" } else { "+" };
871    let accumulator = accumulator(view, &counter_column);
872    format!(
873        "UPDATE {parent_table} SET {counter_column} = {accumulator} {sign} \
874         COALESCE((SELECT {contrib} FROM {child_table} AS {CHILD_ALIAS} \
875           WHERE {CHILD_ALIAS}.{child_pk} = {child_id} \
876             AND {CHILD_ALIAS}.{fk_column} IS NOT NULL{state_predicate}{filter}), 0) \
877         WHERE {parent_table}.{parent_pk} IN \
878           (SELECT {CHILD_ALIAS}.{fk_column} FROM {child_table} AS {CHILD_ALIAS} \
879            WHERE {CHILD_ALIAS}.{child_pk} = {child_id} \
880              AND {CHILD_ALIAS}.{fk_column} \
881                  IS NOT NULL{state_predicate}{filter}{FOR_UPDATE}){tenant}"
882    )
883}
884
885/// How a parent-keyed delta is confined to a tenant.
886#[derive(Debug, Clone, PartialEq, Eq)]
887pub enum TenantScope {
888    /// Require the parent to sit in the same tenant as this child row. Emits no
889    /// predicate for a child model without a tenant column.
890    SameTenantAsChild(i64),
891    /// Require the parent to sit in this tenant, spelled as the text the
892    /// pre-update capture read (`None` for a NULL tenant). Used for the old
893    /// side of a child that changed tenant: its row now says the new tenant,
894    /// so the live predicate would miss the parent it is leaving. Emits no
895    /// predicate for a child model without a tenant column.
896    Captured(Option<String>),
897    /// No tenant predicate. Used only by paths that have no child row to scope
898    /// against (`recompute`, which sweeps the parent table wholesale).
899    Unscoped,
900}
901
902/// Which soft-delete state the child row must be in for a by-id delta to apply.
903#[doc(hidden)]
904#[derive(Debug, Clone, Copy, PartialEq, Eq)]
905pub enum ChildState {
906    /// No `deleted_at` predicate at all.
907    Any,
908    /// Only when the child is live (`deleted_at IS NULL`). Used by the
909    /// decrement, so a repeated soft delete moves nothing.
910    Live,
911    /// Only when the child is soft-deleted. Used by `restore`, so restoring an
912    /// already-live row moves nothing.
913    SoftDeleted,
914}
915
916/// Increment every counter-cached parent of a freshly inserted child record.
917///
918/// A record that arrives already soft-deleted moves nothing: the counter is
919/// defined as live rows only, and every other path agrees on that, so counting a
920/// born-deleted row would inflate the counter until the next repair. An ordinary
921/// insert is live, so this changes nothing for it.
922///
923/// # Errors
924///
925/// Propagates any database error from the `UPDATE`s.
926#[doc(hidden)]
927pub async fn counter_cache_after_insert<M: Send + Sync + 'static>(
928    conn: &mut RuntimeConnection,
929    specs: &[CounterCacheSpec<M>],
930    record: &M,
931) -> AutumnResult<()> {
932    counter_cache_serialize_self_referential(conn, specs).await?;
933    let mut contributions: Vec<Contribution> = Vec::with_capacity(specs.len());
934    for (index, spec) in specs.iter().enumerate() {
935        if !(spec.live_of)(record) {
936            continue;
937        }
938        // A row the derivation's filter rejects contributes 0, and a 0 delta is
939        // no statement at all, not even a `+ 0` write to the parent row.
940        let contrib = (spec.contrib_of)(record);
941        if contrib == 0 {
942            continue;
943        }
944        if let Some(parent_id) = (spec.fk_of)(record) {
945            let scope = TenantScope::SameTenantAsChild((spec.pk_of)(record));
946            contributions.push((index, parent_id, contrib, scope));
947        }
948    }
949    apply_ordered(conn, specs, contributions).await
950}
951
952/// [`counter_cache_after_insert`] over a batch of records (`save_many`).
953///
954/// Children sharing a parent are folded into **one** `+ n` statement per parent,
955/// so inserting a 1000-row chunk issues one `UPDATE` per distinct parent rather
956/// than one per row — still a single atomic statement each, so the concurrency
957/// guarantee is unchanged.
958///
959/// # Errors
960///
961/// Propagates any database error from the `UPDATE`s.
962#[doc(hidden)]
963pub async fn counter_cache_after_insert_many<M: Send + Sync + 'static>(
964    conn: &mut RuntimeConnection,
965    specs: &[CounterCacheSpec<M>],
966    records: &[M],
967) -> AutumnResult<()> {
968    counter_cache_serialize_self_referential(conn, specs).await?;
969    let mut contributions: Vec<Contribution> = Vec::new();
970    for (index, spec) in specs.iter().enumerate() {
971        for record in records {
972            if !(spec.live_of)(record) {
973                continue;
974            }
975            let contrib = (spec.contrib_of)(record);
976            if contrib == 0 {
977                continue;
978            }
979            if let Some(parent_id) = (spec.fk_of)(record) {
980                let scope = TenantScope::SameTenantAsChild((spec.pk_of)(record));
981                contributions.push((index, parent_id, contrib, scope));
982            }
983        }
984    }
985    apply_ordered(conn, specs, contributions).await
986}
987
988/// One parent row a mutation will move: `(spec index, parent id, delta, how
989/// the parent is confined to a tenant)`.
990type Contribution = (usize, i64, i64, TenantScope);
991
992/// One leg's pre-mutation view of a child row.
993#[derive(Debug, Clone, PartialEq, Eq)]
994pub struct Captured {
995    /// The parent the row contributes to.
996    pub parent: i64,
997    /// Its weight in the aggregate. `0` for a row the filter rejects: it *has* a
998    /// parent, and it weighs nothing, which is what makes a filter flip on an
999    /// unchanged parent visible as a delta.
1000    pub contrib: i64,
1001    /// The row's tenant as text, for a leg that reads it (see
1002    /// [`CounterCacheSpec::tenant_of`]); `None` otherwise, or for a NULL tenant.
1003    pub tenant: Option<String>,
1004}
1005
1006/// One leg's pre-mutation [`Captured`] for a child row.
1007///
1008/// `None` when the row contributes to no parent at all: it does not exist, its
1009/// foreign key is NULL, or it is soft-deleted.
1010pub type CapturedContribution = Option<Captured>;
1011
1012/// Every leg's [`CapturedContribution`] for a batch of children, keyed by child
1013/// primary key and sorted by it, as the bulk update paths consume it.
1014pub type CapturedContributions = Vec<(i64, Vec<CapturedContribution>)>;
1015
1016/// Apply every contribution in the lock order all counter-cached mutations
1017/// agree on: `(parent_table, parent_id)`.
1018///
1019/// Ordering by parent id alone is not enough once a mutation moves more than one
1020/// leg. Two child models that declare their legs in opposite orders — one
1021/// `belongs_to(User)` then `belongs_to(Post)`, the other the reverse — would
1022/// take the same two row locks in opposite orders and deadlock, and generated
1023/// repository transactions are single-attempt, so one request fails outright.
1024/// Keying on the parent *table* as well makes the order a property of the schema
1025/// rather than of any one model's declaration order, so every writer in the
1026/// process agrees on it — including two legs that point at the same table, which
1027/// a per-spec ordering cannot reconcile.
1028///
1029/// Folding several children of one parent into a single `+ n` is a pure
1030/// optimization: re-parenting 1000 children becomes two statements rather than
1031/// 2000. It is only sound when the association has no tenant column, since the
1032/// tenant predicate is scoped to a single witness child and a mixed-tenant batch
1033/// folded behind one arbitrary witness would either sweep cross-tenant children
1034/// into the increment or drop legitimate ones. Tenant-scoped specs therefore
1035/// keep one statement per contribution — in the same global order.
1036async fn apply_ordered<M: 'static>(
1037    conn: &mut RuntimeConnection,
1038    specs: &[CounterCacheSpec<M>],
1039    contributions: Vec<Contribution>,
1040) -> AutumnResult<()> {
1041    if contributions.is_empty() {
1042        return Ok(());
1043    }
1044    for (spec_index, parent_id, delta, scope) in fold_and_order(specs, contributions) {
1045        counter_cache_apply_delta(conn, &specs[spec_index], parent_id, delta, scope).await?;
1046    }
1047    Ok(())
1048}
1049
1050/// The pure half of [`apply_ordered`]: fold, drop the no-ops, and put what is
1051/// left in the global lock order. Split out so the ordering is unit-testable
1052/// without a database.
1053fn fold_and_order<M: 'static>(
1054    specs: &[CounterCacheSpec<M>],
1055    contributions: Vec<Contribution>,
1056) -> Vec<Contribution> {
1057    // The running total is an `i128`, so weights that cancel do cancel even
1058    // when no `i64` could hold an intermediate sum (`[MAX, MAX, -MAX, -MAX]`
1059    // is zero, not two statements the database would refuse).
1060    let mut totals: Vec<(usize, i64, i128, TenantScope)> = Vec::with_capacity(contributions.len());
1061    // `(spec, parent)` -> where that parent's running total lives in `totals`.
1062    let mut seen: HashMap<(usize, i64), usize> = HashMap::new();
1063    for (spec_index, parent_id, delta, scope) in contributions {
1064        if specs[spec_index].tenant_column.is_some() {
1065            totals.push((spec_index, parent_id, i128::from(delta), scope));
1066            continue;
1067        }
1068        if let Some(&at) = seen.get(&(spec_index, parent_id)) {
1069            totals[at].2 += i128::from(delta);
1070        } else {
1071            seen.insert((spec_index, parent_id), totals.len());
1072            totals.push((spec_index, parent_id, i128::from(delta), scope));
1073        }
1074    }
1075    // A net total that no single `i64` delta can carry goes out as several
1076    // deltas of the same sign, so the parent moves monotonically from where it
1077    // is to where it ends: when both are representable, so is every step.
1078    // Zero totals are dropped rather than issued as `+ 0`.
1079    let mut folded: Vec<Contribution> = Vec::with_capacity(totals.len());
1080    for (spec_index, parent_id, mut total, scope) in totals {
1081        while total != 0 {
1082            let chunk = total.clamp(i128::from(i64::MIN), i128::from(i64::MAX));
1083            #[allow(clippy::cast_possible_truncation)]
1084            folded.push((spec_index, parent_id, chunk as i64, scope.clone()));
1085            total -= chunk;
1086        }
1087    }
1088    folded
1089        .sort_by_key(|&(spec_index, parent_id, _, _)| (specs[spec_index].parent_table, parent_id));
1090    order_tenant_runs(specs, &mut folded);
1091    folded
1092}
1093
1094/// Order each tenant-scoped parent's deltas so their running sum stays between
1095/// zero and their total.
1096///
1097/// A tenant-scoped leg cannot fold: each delta is guarded by its own child's
1098/// tenant, and two children of one parent need not share one. What it can do
1099/// is choose the order. Walking toward the total and stepping back past it
1100/// keeps every prefix within `[min(0, total), max(0, total)]` whenever the
1101/// deltas allow it at all, so `[MAX, MAX, -MAX]` goes out as `MAX, -MAX, MAX`
1102/// and a parent whose start and end both fit never passes through a value
1103/// that does not. The sort is stable, so the global lock order is unchanged.
1104fn order_tenant_runs<M: 'static>(specs: &[CounterCacheSpec<M>], folded: &mut [Contribution]) {
1105    let mut start = 0;
1106    while start < folded.len() {
1107        let (spec_index, parent_id) = (folded[start].0, folded[start].1);
1108        let mut end = start + 1;
1109        while end < folded.len() && folded[end].0 == spec_index && folded[end].1 == parent_id {
1110            end += 1;
1111        }
1112        if specs[spec_index].tenant_column.is_some() && end - start > 1 {
1113            let run = &mut folded[start..end];
1114            let total: i128 = run.iter().map(|(_, _, delta, _)| i128::from(*delta)).sum();
1115            let mut remaining: Vec<Contribution> = run.to_vec();
1116            let mut prefix: i128 = 0;
1117            for slot in run.iter_mut() {
1118                let toward_total = prefix < total || (prefix == total && total < 0);
1119                // Ties keep their original (child) order, so equal weights go
1120                // out as they were captured and the mutation stays
1121                // deterministic: `min_by_key` takes the first minimum, and the
1122                // reversed scan makes `max_by_key` take the first maximum.
1123                let pick = if toward_total {
1124                    (0..remaining.len()).rev().max_by_key(|&i| remaining[i].2)
1125                } else {
1126                    (0..remaining.len()).min_by_key(|&i| remaining[i].2)
1127                }
1128                .expect("a run has at least one delta left per slot");
1129                let next = remaining.remove(pick);
1130                prefix += i128::from(next.2);
1131                *slot = next;
1132            }
1133        }
1134        start = end;
1135    }
1136}
1137
1138/// Spec indices in the same global order, for the paths that resolve the parent
1139/// inside the statement and so cannot key on its id.
1140///
1141/// This removes every cross-table cycle, which is the half these paths can
1142/// reach: a by-id call touches one parent per leg, so two legs onto *different*
1143/// tables are ordered here, and two legs onto the *same* table are the residual
1144/// (`recompute` is the repair, and a deadlock there surfaces as an error rather
1145/// than as drift). `counter_column` breaks ties so the order is total.
1146fn specs_in_lock_order<M: 'static>(specs: &[CounterCacheSpec<M>]) -> Vec<usize> {
1147    let mut order: Vec<usize> = (0..specs.len()).collect();
1148    order.sort_by_key(|&i| (specs[i].parent_table, specs[i].counter_column));
1149    order
1150}
1151
1152/// Increment every counter-cached parent of the child row `child_id`.
1153///
1154/// The documented escape hatch for applications that insert a child with their
1155/// own SQL inside their own transaction (rather than through the generated
1156/// repository) and still want the framework to own the counter arithmetic:
1157///
1158/// ```rust,ignore
1159/// counter_cache_serialize_self_referential(conn, Comment::counter_caches()).await?;
1160/// let id = diesel::insert_into(comments::table) /* … */ .get_result(conn).await?;
1161/// counter_cache_after_insert_by_id(conn, Comment::counter_caches(), id).await?;
1162/// ```
1163///
1164/// The first line matters on a table with a leg onto itself, and costs nothing
1165/// on any other: every generated mutation takes that advisory lock before its
1166/// first row lock, and an inserted row is locked from the moment it exists (a
1167/// concurrent upsert of the same id waits on it through the primary key), so
1168/// a raw insert that takes the lock only here, after the row, can hold the
1169/// row while a generated mutation holds the lock and waits for the row, and
1170/// the database aborts one of them. Take the lock, then write.
1171///
1172/// For a soft-deleting child the increment is conditional on the row being live,
1173/// mirroring the delete side. Raw SQL can insert a row with `deleted_at` already
1174/// set, and every other path — delete, update, recompute — defines the counter as
1175/// live rows only, so incrementing for a born-deleted row would inflate the
1176/// counter until the next repair.
1177///
1178/// # Errors
1179///
1180/// Propagates any database error from the `UPDATE`s.
1181pub async fn counter_cache_after_insert_by_id<M: 'static>(
1182    conn: &mut RuntimeConnection,
1183    specs: &[CounterCacheSpec<M>],
1184    child_id: i64,
1185) -> AutumnResult<()> {
1186    counter_cache_serialize_self_referential(conn, specs).await?;
1187    for index in specs_in_lock_order(specs) {
1188        let spec = &specs[index];
1189        let state = if spec.child_soft_delete {
1190            ChildState::Live
1191        } else {
1192            ChildState::Any
1193        };
1194        counter_cache_apply_delta_by_child_id(conn, spec, child_id, 1, state).await?;
1195    }
1196    Ok(())
1197}
1198
1199/// Decrement every counter-cached parent of the child row `child_id`.
1200///
1201/// Must be called **before** the row is deleted or soft-deleted: it resolves the
1202/// parent from the still-present child row, and (for a soft-deleting model)
1203/// requires that row to still be live, so the decrement happens exactly once
1204/// however many times the delete is retried.
1205///
1206/// # Errors
1207///
1208/// Propagates any database error from the `UPDATE`s.
1209pub async fn counter_cache_before_delete_by_id<M: 'static>(
1210    conn: &mut RuntimeConnection,
1211    specs: &[CounterCacheSpec<M>],
1212    child_id: i64,
1213) -> AutumnResult<()> {
1214    counter_cache_serialize_self_referential(conn, specs).await?;
1215    for index in specs_in_lock_order(specs) {
1216        let spec = &specs[index];
1217        let state = if spec.child_soft_delete {
1218            ChildState::Live
1219        } else {
1220            ChildState::Any
1221        };
1222        counter_cache_apply_delta_by_child_id(conn, spec, child_id, -1, state).await?;
1223    }
1224    Ok(())
1225}
1226
1227/// Take the child row locks a bulk decrement needs, in ascending id order.
1228///
1229/// The returned rows are discarded — the statement exists for its locks (see
1230/// [`counter_cache_before_delete_many`]'s lock-order note).
1231fn child_lock_sql(view: &SqlView, id_list: &str) -> String {
1232    let Quoted {
1233        child_table,
1234        child_pk,
1235        ..
1236    } = quoted(view);
1237    format!(
1238        "SELECT {child_pk} AS id FROM {child_table} \
1239         WHERE {child_pk} IN ({id_list}) ORDER BY {child_pk}{FOR_UPDATE}"
1240    )
1241}
1242
1243/// [`counter_cache_before_delete_by_id`] over a batch of ids (`delete_many`).
1244///
1245/// Issues **one** statement per spec regardless of batch size: each affected
1246/// parent's counter drops by the number of its children in the batch, computed
1247/// by the database. As with the single-row form, this must run *before* the rows
1248/// are deleted, and for a soft-deleting model it counts only rows that are still
1249/// live so a repeated bulk delete moves nothing.
1250///
1251/// # Lock order
1252///
1253/// The children are locked in ascending id order **before** any parent counter
1254/// is touched. Every other path in this module reaches the parent through a
1255/// child row it has already locked (`counter_cache_apply_delta_by_child_id`
1256/// carries `FOR UPDATE` on its sub-select; the update paths lock the child while
1257/// capturing its foreign keys), so the module's lock order is uniformly
1258/// child-then-parent. Without this statement the bulk path would invert it —
1259/// locking the parent for the decrement here and the child only later, when the
1260/// `DELETE` itself runs — and a bulk delete racing an `update` that re-parents
1261/// one of the same children would deadlock, which generated repository
1262/// transactions do not retry.
1263///
1264/// # Errors
1265///
1266/// Propagates any database error from the `SELECT` or the `UPDATE`s.
1267#[doc(hidden)]
1268pub async fn counter_cache_before_delete_many<M: 'static>(
1269    conn: &mut RuntimeConnection,
1270    specs: &[CounterCacheSpec<M>],
1271    child_ids: &[i64],
1272) -> AutumnResult<()> {
1273    counter_cache_serialize_self_referential(conn, specs).await?;
1274    if specs.is_empty() || child_ids.is_empty() {
1275        return Ok(());
1276    }
1277    let id_list = id_list(child_ids);
1278
1279    // One statement, not one per spec: every spec here is a leg of the *same*
1280    // child model, so they all name the same table and primary key.
1281    let first = view(&specs[0]);
1282    assert_spec_idents(&first);
1283    diesel::sql_query(child_lock_sql(&first, &id_list))
1284        .load::<IdRow>(conn)
1285        .await
1286        .map_err(AutumnError::from)?;
1287
1288    for index in specs_in_lock_order(specs) {
1289        let view = view(&specs[index]);
1290        assert_spec_idents(&view);
1291        refuse_ledgered_parent(&view)?;
1292        let sql = bulk_decrement_sql(&view, &id_list);
1293        diesel::sql_query(sql)
1294            .execute(conn)
1295            .await
1296            .map_err(AutumnError::from)?;
1297    }
1298    Ok(())
1299}
1300
1301/// [`counter_cache_before_delete_many`] restricted to the ONE leg whose foreign
1302/// key is about to be cleared (`dependent = nullify`).
1303///
1304/// A nullify detaches the children from a single parent association; their other
1305/// counter-cached legs are untouched, so decrementing every spec would
1306/// permanently undercount them. For example, nullifying `comments.author_id`
1307/// when a user is deleted must not drop `posts.comment_count` for comments that
1308/// remain attached to their post.
1309///
1310/// # Errors
1311///
1312/// Propagates any database error from the `UPDATE`s.
1313#[doc(hidden)]
1314pub async fn counter_cache_before_detach_many<M: 'static>(
1315    conn: &mut RuntimeConnection,
1316    specs: &[CounterCacheSpec<M>],
1317    fk_column: &str,
1318    child_ids: &[i64],
1319) -> AutumnResult<()> {
1320    counter_cache_serialize_self_referential(conn, specs).await?;
1321    let detached: Vec<CounterCacheSpec<M>> = specs
1322        .iter()
1323        .filter(|spec| spec.fk_column == fk_column)
1324        .copied()
1325        .collect();
1326    counter_cache_before_delete_many(conn, &detached, child_ids).await
1327}
1328
1329/// The one `UPDATE` a bulk decrement issues per leg.
1330///
1331/// The maintained value drops by the aggregate of the batch's qualifying child
1332/// rows, computed by the database in one statement. Both sub-selects carry the
1333/// derivation filter, so a batch of rows the filter rejects matches no parent
1334/// and writes nothing, which is also what keeps NULL arithmetic out of reach.
1335fn bulk_decrement_sql(view: &SqlView, id_list: &str) -> String {
1336    let Quoted {
1337        child_table,
1338        child_pk,
1339        fk_column,
1340        parent_table,
1341        parent_pk,
1342        counter_column,
1343        ..
1344    } = quoted(view);
1345    let live = if view.child_soft_delete {
1346        live_predicate(view, true)
1347    } else {
1348        String::new()
1349    };
1350    let filter = filter_predicate(view, CHILD_ALIAS);
1351    let aggregate = aggregate_expr(view, CHILD_ALIAS);
1352    // Both sub-selects correlate on the parent, so the tenant check is a
1353    // plain column comparison inside each: no extra round trip, and a
1354    // cross-tenant child contributes to neither the count nor the row set.
1355    let tenant = tenant_predicate_joined(view);
1356    let accumulator = accumulator(view, &counter_column);
1357    let (minuend, aggregate) = bulk_decrement_parts(view, &accumulator, aggregate);
1358    format!(
1359        "UPDATE {parent_table} SET {counter_column} = {minuend}\
1360         (SELECT {aggregate} FROM {child_table} AS {CHILD_ALIAS} \
1361          WHERE {CHILD_ALIAS}.{fk_column} = {parent_table}.{parent_pk} \
1362            AND {CHILD_ALIAS}.{child_pk} IN ({id_list}){live}{filter}{tenant}) \
1363         WHERE {parent_table}.{parent_pk} IN \
1364           (SELECT {CHILD_ALIAS}.{fk_column} FROM {child_table} AS {CHILD_ALIAS} \
1365            WHERE {CHILD_ALIAS}.{child_pk} IN ({id_list}) \
1366              AND {CHILD_ALIAS}.{fk_column} IS NOT NULL{live}{filter} \
1367              AND {CHILD_ALIAS}.{fk_column} = {parent_table}.{parent_pk}{tenant})"
1368    )
1369}
1370
1371/// The text before the bulk decrement's sub-select, and its select list.
1372///
1373/// `<current> - ` and the batch's aggregate.
1374#[cfg(not(feature = "sqlite"))]
1375fn bulk_decrement_parts(_view: &SqlView, current: &str, aggregate: String) -> (String, String) {
1376    (format!("{current} - "), aggregate)
1377}
1378
1379/// The text before the bulk decrement's sub-select, and its select list.
1380///
1381/// A `SQLite` sum moves the parent's value into the sub-select, and joins it
1382/// with the batch's split sums (#2663). The batch's own sum can be outside
1383/// `i64` when the parent's new value is in range: deleting `MAX, MAX` from
1384/// `MAX, MAX, MIN` gives `MIN`. A count cannot overflow, so it keeps the
1385/// plain form.
1386#[cfg(feature = "sqlite")]
1387fn bulk_decrement_parts(view: &SqlView, current: &str, aggregate: String) -> (String, String) {
1388    if view.counts_rows() {
1389        return (format!("{current} - "), aggregate);
1390    }
1391    // Qualified: inside the sub-select, a bare name can bind to the child.
1392    let Quoted {
1393        parent_table,
1394        counter_column,
1395        ..
1396    } = quoted(view);
1397    let parent_value = accumulator(view, &format!("{parent_table}.{counter_column}"));
1398    let (hi, lo) = split_sums(&contrib_expr(view, CHILD_ALIAS));
1399    let hi = format!("(({parent_value}) >> 32) - {hi}");
1400    let lo = format!("(({parent_value}) & {LOW_MASK}) - {lo}");
1401    (String::new(), split_join(&hi, &lo))
1402}
1403
1404/// Render `ids` as a literal SQL list.
1405///
1406/// The values are `i64`, so their decimal rendering cannot contain anything a
1407/// SQL parser would treat as syntax — this is a type-level guarantee, not an
1408/// escaping convention. Inlining them (rather than binding) keeps the statement
1409/// backend-portable: `SQLite` has no array type for a bound `= ANY($1)`.
1410fn id_list(ids: &[i64]) -> String {
1411    let mut out = String::new();
1412    for (i, id) in ids.iter().enumerate() {
1413        if i > 0 {
1414            out.push(',');
1415        }
1416        out.push_str(&id.to_string());
1417    }
1418    out
1419}
1420
1421/// Increment every counter-cached parent of a child being **restored** from a
1422/// soft delete.
1423///
1424/// Must be called **before** `deleted_at` is cleared: the delta only applies
1425/// while the row is still soft-deleted, so restoring an already-live row is a
1426/// no-op rather than an inflation.
1427///
1428/// # Errors
1429///
1430/// Propagates any database error from the `UPDATE`s.
1431#[doc(hidden)]
1432pub async fn counter_cache_before_restore_by_id<M: 'static>(
1433    conn: &mut RuntimeConnection,
1434    specs: &[CounterCacheSpec<M>],
1435    child_id: i64,
1436) -> AutumnResult<()> {
1437    counter_cache_serialize_self_referential(conn, specs).await?;
1438    for index in specs_in_lock_order(specs) {
1439        let spec = &specs[index];
1440        // A model with no `deleted_at` has no restore path; guard anyway so a
1441        // mis-wired call cannot double-increment.
1442        if spec.child_soft_delete {
1443            counter_cache_apply_delta_by_child_id(conn, spec, child_id, 1, ChildState::SoftDeleted)
1444                .await?;
1445        }
1446    }
1447    Ok(())
1448}
1449
1450/// Read each counter-cached leg's current foreign key **and contribution** for
1451/// child `child_id`.
1452///
1453/// Called **before** an update so the post-update record can be compared against
1454/// it; the outer `Option` is `None` when the child row does not exist, has no
1455/// parent, or is soft-deleted. Issues no statement at all when the model has no
1456/// counter caches.
1457///
1458/// The contribution is read here rather than recomputed later because the
1459/// pre-update row is what it is a function of, and that row is gone once the
1460/// `UPDATE` lands. A row the filter rejects reports `0`, which is what makes a
1461/// filter flip on an unchanged parent a `+1` rather than a no-op.
1462///
1463/// # Errors
1464///
1465/// Propagates any database error from the `SELECT`s.
1466#[doc(hidden)]
1467pub async fn counter_cache_capture_fks<M: 'static>(
1468    conn: &mut RuntimeConnection,
1469    specs: &[CounterCacheSpec<M>],
1470    child_id: i64,
1471) -> AutumnResult<Vec<CapturedContribution>> {
1472    counter_cache_serialize_self_referential(conn, specs).await?;
1473    if specs.is_empty() {
1474        return Ok(Vec::new());
1475    }
1476    let mut out = Vec::with_capacity(specs.len());
1477    for spec in specs {
1478        let view = view(spec);
1479        assert_spec_idents(&view);
1480        let Quoted {
1481            child_table,
1482            child_pk,
1483            fk_column,
1484            ..
1485        } = quoted(&view);
1486        // A soft-deleted child is counted by nobody, so it has no "old parent"
1487        // to move away from — reporting one would make a later re-parent
1488        // decrement a counter that had already dropped this row.
1489        let live = live_predicate(&view, true);
1490        // A leg that reads the tenant captures it as well, as text: the old
1491        // side of a tenant change is guarded by this value, not by the row's
1492        // (by then new) tenant.
1493        if let Some(tenant_column) = captured_tenant_column(spec, &view) {
1494            let contrib = contrib_case_expr(&view, CHILD_ALIAS);
1495            let sql = format!(
1496                "SELECT {CHILD_ALIAS}.{fk_column} AS fk_value, \
1497                 {contrib} AS contrib_value, \
1498                 CAST({CHILD_ALIAS}.{tenant_column} AS TEXT) AS tenant_text \
1499                 FROM {child_table} AS {CHILD_ALIAS} \
1500                 WHERE {CHILD_ALIAS}.{child_pk} = {PH1}{live}{FOR_UPDATE}"
1501            );
1502            let row: Option<FkContribTenantRow> = diesel::sql_query(sql)
1503                .bind::<BigInt, _>(child_id)
1504                .get_result::<FkContribTenantRow>(conn)
1505                .await
1506                .optional_row()?;
1507            out.push(row.and_then(FkContribTenantRow::captured));
1508            continue;
1509        }
1510        if view.is_plain() {
1511            let sql = format!(
1512                "SELECT {CHILD_ALIAS}.{fk_column} AS fk_value \
1513                 FROM {child_table} AS {CHILD_ALIAS} \
1514                 WHERE {CHILD_ALIAS}.{child_pk} = {PH1}{live}{FOR_UPDATE}"
1515            );
1516            let row: Option<FkRow> = diesel::sql_query(sql)
1517                .bind::<BigInt, _>(child_id)
1518                .get_result::<FkRow>(conn)
1519                .await
1520                .optional_row()?;
1521            out.push(row.and_then(|r| r.fk_value).map(|fk| captured(fk, 1)));
1522            continue;
1523        }
1524        let contrib = contrib_case_expr(&view, CHILD_ALIAS);
1525        let sql = format!(
1526            "SELECT {CHILD_ALIAS}.{fk_column} AS fk_value, \
1527             {contrib} AS contrib_value \
1528             FROM {child_table} AS {CHILD_ALIAS} \
1529             WHERE {CHILD_ALIAS}.{child_pk} = {PH1}{live}{FOR_UPDATE}"
1530        );
1531        let row: Option<FkContribRow> = diesel::sql_query(sql)
1532            .bind::<BigInt, _>(child_id)
1533            .get_result::<FkContribRow>(conn)
1534            .await
1535            .optional_row()?;
1536        out.push(row.and_then(|r| r.fk_value.map(|fk| captured(fk, r.contrib_value))));
1537    }
1538    Ok(out)
1539}
1540
1541/// A [`Captured`] with no tenant, for the legs that do not read one.
1542const fn captured(parent: i64, contrib: i64) -> Captured {
1543    Captured {
1544        parent,
1545        contrib,
1546        tenant: None,
1547    }
1548}
1549
1550/// The quoted tenant column the capture reads, when this leg reads one: it
1551/// has a tenant column *and* an accessor for the new side to compare against.
1552/// Without the accessor the capture stays as it was, so a leg that cannot see
1553/// its tenant costs nothing extra.
1554fn captured_tenant_column<M: 'static>(
1555    spec: &CounterCacheSpec<M>,
1556    view: &SqlView,
1557) -> Option<String> {
1558    let tenant_column = view.tenant_column?;
1559    spec.tenant_of?;
1560    Some(quote_ident(tenant_column))
1561}
1562
1563/// Move the counters for a child whose foreign keys may have just changed.
1564///
1565/// `before` is the slice [`counter_cache_capture_fks`] returned prior to the
1566/// update; `record` is the row as persisted after it. A leg whose foreign key is
1567/// unchanged issues **no** statement — an ordinary field edit costs nothing.
1568///
1569/// # Errors
1570///
1571/// Propagates any database error from the `UPDATE`s.
1572#[doc(hidden)]
1573pub async fn counter_cache_after_update<M: Send + Sync + 'static>(
1574    conn: &mut RuntimeConnection,
1575    specs: &[CounterCacheSpec<M>],
1576    before: &[CapturedContribution],
1577    record: &M,
1578) -> AutumnResult<()> {
1579    let mut moves: Vec<Contribution> = Vec::with_capacity(specs.len() * 2);
1580    for (index, spec) in specs.iter().enumerate() {
1581        let old = before.get(index).cloned().flatten();
1582        let new = contribution_of(spec, record);
1583        // Collected rather than applied here, and not old-then-new: every delta
1584        // this mutation makes goes out in one global lock order (see
1585        // `apply_ordered`), so two transactions swapping children between the
1586        // same parents cannot take the two row locks in opposite orders.
1587        push_diff(index, old, new, (spec.pk_of)(record), &mut moves);
1588    }
1589    apply_ordered(conn, specs, moves).await
1590}
1591
1592/// A record's `(parent, contribution)` after a mutation, or `None` when it
1593/// contributes to no parent.
1594///
1595/// A soft-deleted child is counted by nobody, so neither its old nor its new
1596/// parent may move. The generated `update` does not filter soft-deleted rows, so
1597/// this is reachable.
1598fn contribution_of<M: 'static>(spec: &CounterCacheSpec<M>, record: &M) -> CapturedContribution {
1599    if !(spec.live_of)(record) {
1600        return None;
1601    }
1602    (spec.fk_of)(record).map(|parent| Captured {
1603        parent,
1604        contrib: (spec.contrib_of)(record),
1605        tenant: spec.tenant_of.and_then(|tenant_of| tenant_of(record)),
1606    })
1607}
1608
1609/// Turn one leg's before/after [`Captured`] into deltas.
1610///
1611/// Unchanged ⇒ nothing. Same parent and tenant, different weight (a filter
1612/// flip, or an edited summed field) ⇒ one delta for the difference. Different
1613/// parent, or different tenant ⇒ the old weight off the old parent and the new
1614/// weight onto the new one, each skipped when it is 0, so a row the filter
1615/// rejects never touches a parent row.
1616///
1617/// The new side is always scoped by the child row as it now is. The old side
1618/// is too, unless the tenant changed: then the row no longer says which tenant
1619/// the old parent sits in, so the removal is scoped by the tenant the capture
1620/// read, and a parent left behind in the old tenant is found rather than
1621/// missed.
1622fn push_diff(
1623    index: usize,
1624    old: CapturedContribution,
1625    new: CapturedContribution,
1626    witness: i64,
1627    out: &mut Vec<Contribution>,
1628) {
1629    if old == new {
1630        return;
1631    }
1632    let live = TenantScope::SameTenantAsChild(witness);
1633    let new_tenant = new.as_ref().and_then(|new| new.tenant.clone());
1634    if let (Some(old), Some(new)) = (&old, &new)
1635        && old.parent == new.parent
1636        && old.tenant == new.tenant
1637    {
1638        // Two representable weights need not have a representable difference
1639        // (`i64::MIN` edited to `i64::MAX`). Then the move goes out as the old
1640        // weight off and the new weight on, which reach the same total.
1641        if let Some(delta) = new.contrib.checked_sub(old.contrib) {
1642            if delta != 0 {
1643                out.push((index, old.parent, delta, live));
1644            }
1645            return;
1646        }
1647        push_removal(index, old.parent, old.contrib, live.clone(), out);
1648        out.push((index, new.parent, new.contrib, live));
1649        return;
1650    }
1651    if let Some(old) = old
1652        && old.contrib != 0
1653    {
1654        let scope = if old.tenant == new_tenant {
1655            live.clone()
1656        } else {
1657            TenantScope::Captured(old.tenant)
1658        };
1659        push_removal(index, old.parent, old.contrib, scope, out);
1660    }
1661    if let Some(new) = new
1662        && new.contrib != 0
1663    {
1664        out.push((index, new.parent, new.contrib, live));
1665    }
1666}
1667
1668/// Take `contrib` off `parent`.
1669///
1670/// `-i64::MIN` is not an `i64`, so that one weight leaves as two halves that
1671/// are. Every other weight negates in place.
1672fn push_removal(
1673    index: usize,
1674    parent: i64,
1675    contrib: i64,
1676    scope: TenantScope,
1677    out: &mut Vec<Contribution>,
1678) {
1679    if let Some(negated) = contrib.checked_neg() {
1680        out.push((index, parent, negated, scope));
1681    } else {
1682        let half = contrib / 2;
1683        out.push((index, parent, -half, scope.clone()));
1684        out.push((index, parent, -(contrib - half), scope));
1685    }
1686}
1687
1688/// [`counter_cache_capture_fks`] over a batch of child ids (`update_many`).
1689///
1690/// Issues one `SELECT` per spec (not per id), returning `(child id,
1691/// (parent, contribution) in spec order)` for every row found. Rows absent from
1692/// the table are simply absent from the result.
1693///
1694/// # Errors
1695///
1696/// Propagates any database error from the `SELECT`s.
1697#[doc(hidden)]
1698pub async fn counter_cache_capture_fks_many<M: 'static>(
1699    conn: &mut RuntimeConnection,
1700    specs: &[CounterCacheSpec<M>],
1701    child_ids: &[i64],
1702) -> AutumnResult<CapturedContributions> {
1703    counter_cache_serialize_self_referential(conn, specs).await?;
1704    if specs.is_empty() || child_ids.is_empty() {
1705        return Ok(Vec::new());
1706    }
1707    let id_list = id_list(child_ids);
1708    let mut by_child: HashMap<i64, Vec<CapturedContribution>> = HashMap::new();
1709    for (index, spec) in specs.iter().enumerate() {
1710        let view = view(spec);
1711        assert_spec_idents(&view);
1712        let Quoted {
1713            child_table,
1714            child_pk,
1715            fk_column,
1716            ..
1717        } = quoted(&view);
1718        let live = live_predicate(&view, true);
1719        if let Some(tenant_column) = captured_tenant_column(spec, &view) {
1720            let contrib = contrib_case_expr(&view, CHILD_ALIAS);
1721            let sql = format!(
1722                "SELECT {CHILD_ALIAS}.{child_pk} AS child_id, \
1723                 {CHILD_ALIAS}.{fk_column} AS fk_value, \
1724                 {contrib} AS contrib_value, \
1725                 CAST({CHILD_ALIAS}.{tenant_column} AS TEXT) AS tenant_text \
1726                 FROM {child_table} AS {CHILD_ALIAS} \
1727                 WHERE {CHILD_ALIAS}.{child_pk} IN ({id_list}){live} \
1728                 ORDER BY {CHILD_ALIAS}.{child_pk}{FOR_UPDATE}"
1729            );
1730            let rows: Vec<ChildFkContribTenantRow> = diesel::sql_query(sql)
1731                .load::<ChildFkContribTenantRow>(conn)
1732                .await
1733                .map_err(AutumnError::from)?;
1734            for row in rows {
1735                let entry = by_child
1736                    .entry(row.child_id)
1737                    .or_insert_with(|| vec![None; specs.len()]);
1738                entry[index] = row.captured();
1739            }
1740            continue;
1741        }
1742        if view.is_plain() {
1743            let sql = format!(
1744                "SELECT {CHILD_ALIAS}.{child_pk} AS child_id, \
1745                 {CHILD_ALIAS}.{fk_column} AS fk_value \
1746                 FROM {child_table} AS {CHILD_ALIAS} \
1747                 WHERE {CHILD_ALIAS}.{child_pk} IN ({id_list}){live} \
1748                 ORDER BY {CHILD_ALIAS}.{child_pk}{FOR_UPDATE}"
1749            );
1750            let rows: Vec<ChildFkRow> = diesel::sql_query(sql)
1751                .load::<ChildFkRow>(conn)
1752                .await
1753                .map_err(AutumnError::from)?;
1754            for row in rows {
1755                let entry = by_child
1756                    .entry(row.child_id)
1757                    .or_insert_with(|| vec![None; specs.len()]);
1758                entry[index] = row.fk_value.map(|fk| captured(fk, 1));
1759            }
1760            continue;
1761        }
1762        let contrib = contrib_case_expr(&view, CHILD_ALIAS);
1763        let sql = format!(
1764            "SELECT {CHILD_ALIAS}.{child_pk} AS child_id, \
1765             {CHILD_ALIAS}.{fk_column} AS fk_value, \
1766             {contrib} AS contrib_value \
1767             FROM {child_table} AS {CHILD_ALIAS} \
1768             WHERE {CHILD_ALIAS}.{child_pk} IN ({id_list}){live} \
1769             ORDER BY {CHILD_ALIAS}.{child_pk}{FOR_UPDATE}"
1770        );
1771        let rows: Vec<ChildFkContribRow> = diesel::sql_query(sql)
1772            .load::<ChildFkContribRow>(conn)
1773            .await
1774            .map_err(AutumnError::from)?;
1775        for row in rows {
1776            let entry = by_child
1777                .entry(row.child_id)
1778                .or_insert_with(|| vec![None; specs.len()]);
1779            entry[index] = row.fk_value.map(|fk| captured(fk, row.contrib_value));
1780        }
1781    }
1782    let mut out: CapturedContributions = by_child.into_iter().collect();
1783    out.sort_unstable_by_key(|(id, _)| *id);
1784    Ok(out)
1785}
1786
1787/// [`counter_cache_after_update`] over a batch of post-update records
1788/// (`update_many`), matched back to their pre-update foreign keys by primary key.
1789///
1790/// A record with no captured "before" entry (its row appeared between the
1791/// capture and the update) is skipped rather than guessed at; the recompute
1792/// repair path exists for exactly that kind of edge.
1793///
1794/// # Errors
1795///
1796/// Propagates any database error from the `UPDATE`s.
1797#[doc(hidden)]
1798pub async fn counter_cache_after_update_many<M: Send + Sync + 'static>(
1799    conn: &mut RuntimeConnection,
1800    specs: &[CounterCacheSpec<M>],
1801    before: &[(i64, Vec<CapturedContribution>)],
1802    records: &[M],
1803) -> AutumnResult<()> {
1804    if specs.is_empty() {
1805        return Ok(());
1806    }
1807    let pk_of = specs[0].pk_of;
1808    let mut contributions: Vec<Contribution> = Vec::new();
1809    for (index, spec) in specs.iter().enumerate() {
1810        // Every leg's deltas are collected before any statement runs, then
1811        // folded and applied in one global lock order (see `apply_ordered`).
1812        // Re-parenting 1000 children from one post to another is then two
1813        // `UPDATE`s (`-1000`, `+1000`) rather than 2000, and two concurrent
1814        // batches swapping children between the same parents take their row
1815        // locks in the same order and cannot deadlock.
1816        for record in records {
1817            let child_id = pk_of(record);
1818            let Ok(found) = before.binary_search_by_key(&child_id, |(id, _)| *id) else {
1819                // No captured "before" for this row: it appeared between the
1820                // capture and the update. Guessing would be worse than leaving
1821                // it to `recompute`.
1822                continue;
1823            };
1824            let old = before[found].1.get(index).cloned().flatten();
1825            let new = contribution_of(spec, record);
1826            push_diff(index, old, new, child_id, &mut contributions);
1827        }
1828    }
1829    apply_ordered(conn, specs, contributions).await
1830}
1831
1832/// Move the counters for an `upsert_many` chunk (#1325).
1833///
1834/// `existing` is the pre-upsert snapshot the upsert already loaded (and locked)
1835/// for this chunk, so no extra query is needed: a record present in it is an
1836/// **update** and gets the before/after foreign-key diff; a record absent from it
1837/// is an **insert** and gets a plain increment.
1838///
1839/// # The one row this classification can get wrong
1840///
1841/// The snapshot is loaded `FOR UPDATE`, so every row that *existed* when it ran
1842/// is locked: it cannot be re-parented underneath this chunk, and its diff is
1843/// exact. The gap is the row that did **not** exist then — there is nothing to
1844/// lock — and that another transaction inserts before this chunk's `INSERT …
1845/// ON CONFLICT` runs. Postgres then updates that row rather than inserting it,
1846/// while this classifies it as an insert: the `+1` duplicates the increment the
1847/// other transaction already made, and if the upsert also moved the foreign key,
1848/// the old parent never gets its decrement.
1849///
1850/// Closing it needs the discrimination to come from the statement that did the
1851/// work (Postgres exposes it as `xmax = 0` in `RETURNING`), because no
1852/// pre-upsert read can see a row that does not exist yet — row locks do not
1853/// block inserts, and the advisory lock the *versioned* upsert takes only
1854/// serializes upserts against each other, not against a plain `save` or a raw
1855/// insert. It is drift, not corruption, and `recompute` repairs it.
1856///
1857/// `SQLite` is unaffected: `BEGIN IMMEDIATE` excludes every other writer for the
1858/// duration of the transaction, so the window does not exist there.
1859///
1860/// # Errors
1861///
1862/// Propagates any database error from the `UPDATE`s.
1863#[doc(hidden)]
1864pub async fn counter_cache_after_upsert_many<M: Send + Sync + 'static>(
1865    conn: &mut RuntimeConnection,
1866    specs: &[CounterCacheSpec<M>],
1867    existing: &[M],
1868    upserted: &[M],
1869) -> AutumnResult<()> {
1870    counter_cache_serialize_self_referential(conn, specs).await?;
1871    if specs.is_empty() {
1872        return Ok(());
1873    }
1874    let pk_of = specs[0].pk_of;
1875    // A soft-deleted existing row is counted by nobody, so it has no "old
1876    // parent". Recording one would make the post-upsert side — which now checks
1877    // `live_of` and yields `None` for a row that stays deleted — decrement a
1878    // parent that had already dropped it.
1879    let before: HashMap<i64, Vec<CapturedContribution>> = existing
1880        .iter()
1881        .map(|row| {
1882            (
1883                pk_of(row),
1884                specs
1885                    .iter()
1886                    .map(|spec| contribution_of(spec, row))
1887                    .collect(),
1888            )
1889        })
1890        .collect();
1891    // One pass per leg over the whole chunk, inserts and updates together: a row
1892    // absent from `before` is an insert (`+1`), a row present in it is an update
1893    // (the before/after diff). Folded and ordered by `apply_ordered` like every
1894    // other bulk path, for the same two reasons.
1895    let mut contributions: Vec<Contribution> = Vec::new();
1896    for (index, spec) in specs.iter().enumerate() {
1897        for record in upserted {
1898            let child_id = pk_of(record);
1899            let new = contribution_of(spec, record);
1900            match before.get(&child_id) {
1901                None => {
1902                    if let Some(new) = new
1903                        && new.contrib != 0
1904                    {
1905                        let scope = TenantScope::SameTenantAsChild(child_id);
1906                        contributions.push((index, new.parent, new.contrib, scope));
1907                    }
1908                }
1909                Some(old_fks) => {
1910                    let old = old_fks.get(index).cloned().flatten();
1911                    push_diff(index, old, new, child_id, &mut contributions);
1912                }
1913            }
1914        }
1915    }
1916    apply_ordered(conn, specs, contributions).await
1917}
1918
1919/// How many parent rows one recompute transaction locks and repairs at a time.
1920///
1921/// A full sweep has to take a row lock on every parent it rebuilds (see
1922/// [`recompute_batch`]), and locks are held until commit. Rebuilding the whole
1923/// table in one transaction would therefore block every concurrent write to that
1924/// table for the duration of the sweep, so the sweep is cut into batches: each
1925/// one is its own short transaction, and per-parent correctness does not depend
1926/// on the batches being atomic with each other.
1927const RECOMPUTE_BATCH: i64 = 1_000;
1928
1929/// The maintained value as the source of truth defines it, correlated on the
1930/// parent row.
1931///
1932/// The correlated sub-select aliases the child table so a self-referential
1933/// counter cache (child table == parent table) still binds the outer statement's
1934/// target on the right-hand side of the join predicate.
1935///
1936/// The ground truth has to agree with what the deltas maintain: an ordinary
1937/// delta skips a cross-tenant child and a filtered-out row, so a repair that
1938/// counted either would undo the isolation, or the filter, on the very next
1939/// sweep.
1940pub fn ground_truth_sql(view: &SqlView) -> String {
1941    let Quoted {
1942        child_table,
1943        fk_column,
1944        parent_table,
1945        parent_pk,
1946        ..
1947    } = quoted(view);
1948    let live = live_predicate(view, true);
1949    let filter = filter_predicate(view, CHILD_ALIAS);
1950    let tenant = tenant_predicate_joined(view);
1951    let aggregate = aggregate_expr(view, CHILD_ALIAS);
1952    format!(
1953        "(SELECT {aggregate} FROM {child_table} AS {CHILD_ALIAS} \
1954          WHERE {CHILD_ALIAS}.{fk_column} = {parent_table}.{parent_pk}{live}{filter}{tenant})"
1955    )
1956}
1957
1958/// The `UPDATE` that rebuilds `ids`' maintained values from the source of truth.
1959pub fn recompute_update_sql(view: &SqlView, ids: &str) -> String {
1960    let Quoted {
1961        parent_table,
1962        parent_pk,
1963        counter_column,
1964        ..
1965    } = quoted(view);
1966    let ground_truth = ground_truth_sql(view);
1967    // `IS DISTINCT FROM` so a sweep over a healthy table writes nothing: under
1968    // MVCC an unconditional assignment would rewrite every parent row (bloat
1969    // proportional to the whole table, for no change), and it would make the
1970    // returned count the row count rather than the repair count.
1971    format!(
1972        "UPDATE {parent_table} SET {counter_column} = {ground_truth} \
1973         WHERE {parent_table}.{parent_pk} IN ({ids}) \
1974           AND {parent_table}.{counter_column} {IS_DISTINCT_FROM} {ground_truth}"
1975    )
1976}
1977
1978/// How many of the newest `limit` parent rows disagree with the source of
1979/// truth.
1980///
1981/// One aggregate statement per derivation, so it is a single round trip. The
1982/// `LIMIT` bounds the **input**: the `limit` highest-id parents are the window,
1983/// materialised under the parent table's own name so the correlated ground
1984/// truth needs no rewriting, and only those rows pay for a correlated
1985/// aggregate. A `LIMIT` on the matching output alone would not do it: on a
1986/// healthy table nothing matches, so the engine would evaluate the aggregate
1987/// for every parent before proving there is nothing left to find. The question
1988/// an operator asks is "is this derivation drifting", not "by exactly how much
1989/// on a billion-row table", and newest-first is where live writes, and so live
1990/// drift, concentrate.
1991pub fn drift_sql(view: &SqlView, limit: i64) -> String {
1992    let Quoted {
1993        parent_table,
1994        parent_pk,
1995        counter_column,
1996        ..
1997    } = quoted(view);
1998    let ground_truth = ground_truth_sql(view);
1999    format!(
2000        "SELECT COUNT(*) AS count FROM \
2001         (SELECT 1 AS drifted FROM \
2002           (SELECT * FROM {parent_table} ORDER BY {parent_table}.{parent_pk} DESC LIMIT {limit}) \
2003           AS {parent_table} \
2004          WHERE {parent_table}.{counter_column} {IS_DISTINCT_FROM} {ground_truth}) \
2005         AS __autumn_cc_drift"
2006    )
2007}
2008
2009/// Repair one batch of parents: lock them, then rebuild them, in one transaction.
2010///
2011/// The lock is what makes a repair safe to run against live traffic. Without it,
2012/// on Postgres, a repair racing an in-flight child insert **introduces** the
2013/// drift it exists to remove: the child's transaction has already taken the
2014/// parent's row lock for its `SET c = c + 1` but has not committed, so the
2015/// repair's `UPDATE` blocks on that lock, and when it resumes it writes a
2016/// `COUNT(*)` taken from a snapshot that predates — and therefore cannot see —
2017/// the child that just committed. The committed `+1` is silently overwritten.
2018///
2019/// Taking the lock in a *separate, earlier statement* removes both halves of
2020/// that race. Either the repair wins the lock, in which case the child's
2021/// increment is applied after the repair commits and is relative, so it lands on
2022/// top of the rebuilt value; or the child wins, in which case the locking
2023/// `SELECT` waits for it to commit and the `UPDATE` that follows takes a fresh
2024/// snapshot that does see it. Ids are locked in ascending order, matching the
2025/// order the delta paths apply in, so the two cannot deadlock against each
2026/// other.
2027///
2028/// On `SQLite` the enclosing `BEGIN IMMEDIATE` already excludes every other
2029/// writer, and `FOR UPDATE` degrades to the empty string; the locking `SELECT`
2030/// is then a cheap indexed read that keeps this one code path.
2031async fn recompute_batch(
2032    conn: &mut RuntimeConnection,
2033    view: &SqlView,
2034    ids: &[i64],
2035) -> AutumnResult<usize> {
2036    if ids.is_empty() {
2037        return Ok(0);
2038    }
2039    let view = *view;
2040    let ids = ids.to_vec();
2041    scoped_immediate_transaction::<usize, AutumnError, _>(conn, move |conn| {
2042        async move { recompute_batch_statements(conn, &view, &ids).await }.scope_boxed()
2043    })
2044    .await
2045}
2046
2047/// The two statements [`recompute_batch`] runs, without the transaction.
2048///
2049/// Split out because the derivation backfill has to commit the repaired batch
2050/// and its checkpoint **together**: it opens one transaction and runs this plus
2051/// the checkpoint write inside it. A checkpoint committed separately from the
2052/// batch it describes would double-apply or skip a batch after a crash.
2053pub async fn recompute_batch_statements(
2054    conn: &mut RuntimeConnection,
2055    view: &SqlView,
2056    ids: &[i64],
2057) -> AutumnResult<usize> {
2058    if ids.is_empty() {
2059        return Ok(0);
2060    }
2061    assert_spec_idents(view);
2062    refuse_ledgered_parent(view)?;
2063    let id_list = id_list(ids);
2064    let parent_table = quote_ident(view.parent_table);
2065    let parent_pk = quote_ident(view.parent_pk);
2066    let lock_sql = format!(
2067        "SELECT {parent_pk} AS id FROM {parent_table} \
2068         WHERE {parent_pk} IN ({id_list}) ORDER BY {parent_pk}{FOR_UPDATE}"
2069    );
2070    diesel::sql_query(lock_sql)
2071        .load::<IdRow>(&mut *conn)
2072        .await
2073        .map_err(AutumnError::from)?;
2074    diesel::sql_query(recompute_update_sql(view, &id_list))
2075        .execute(&mut *conn)
2076        .await
2077        .map_err(AutumnError::from)
2078}
2079
2080/// One page of parent primary keys after `cursor`, in ascending order.
2081///
2082/// A parent inserted after its page was read is simply not in this sweep, which
2083/// is harmless: it starts at the column default and its children are counted by
2084/// the delta paths.
2085///
2086/// [`recompute_view`] reads pages outside its repair transactions, so no lock is
2087/// held while enumerating. The derivation backfill
2088/// ([`crate::derivation::run_backfill`]) reads its page **inside** the batch
2089/// transaction instead, because the cursor it pages from is the checkpoint on
2090/// the state row that transaction holds the lock on. Reading it outside would
2091/// mean paging from a cursor another replica had already moved.
2092pub async fn parent_id_page(
2093    conn: &mut RuntimeConnection,
2094    view: &SqlView,
2095    cursor: Option<i64>,
2096    limit: i64,
2097) -> AutumnResult<Vec<i64>> {
2098    assert_spec_idents(view);
2099    let parent_table = quote_ident(view.parent_table);
2100    let parent_pk = quote_ident(view.parent_pk);
2101    let mut page_sql = format!("SELECT {parent_pk} AS id FROM {parent_table}");
2102    if cursor.is_some() {
2103        let _ = write!(page_sql, " WHERE {parent_pk} > {PH1}");
2104    }
2105    let _ = write!(page_sql, " ORDER BY {parent_pk} LIMIT {limit}");
2106    let query = diesel::sql_query(page_sql);
2107    let page = if let Some(after) = cursor {
2108        query.bind::<BigInt, _>(after).load::<IdRow>(conn).await
2109    } else {
2110        query.load::<IdRow>(conn).await
2111    }
2112    .map_err(AutumnError::from)?;
2113    Ok(page.iter().map(|row| row.id).collect())
2114}
2115
2116/// Rebuild one maintained value from the source of truth, one batch at a time.
2117///
2118/// The non-generic core of [`counter_cache_recompute`], so the derivation
2119/// repair path ([`crate::derivation::recompute`]) runs exactly the same sweep.
2120pub async fn recompute_view(
2121    conn: &mut RuntimeConnection,
2122    view: &SqlView,
2123    parent_id: Option<i64>,
2124) -> AutumnResult<usize> {
2125    assert_spec_idents(view);
2126    if let Some(id) = parent_id {
2127        return recompute_batch(conn, view, &[id]).await;
2128    }
2129    // A self-referential table (a comment's `reply_count`) has rows that are
2130    // children and parents at once: a batch holding several parents in id
2131    // order could form a lock cycle with a mutation that holds a child row and
2132    // wants its parent. One parent per transaction takes one lock and never a
2133    // second, so the repair cannot be one side of that cycle.
2134    let batch = if view.child_table == view.parent_table {
2135        1
2136    } else {
2137        RECOMPUTE_BATCH
2138    };
2139    let mut touched = 0usize;
2140    let mut cursor: Option<i64> = None;
2141    loop {
2142        let ids = parent_id_page(conn, view, cursor, batch).await?;
2143        let Some(&last) = ids.last() else { break };
2144        cursor = Some(last);
2145        touched += recompute_batch_retrying(conn, view, &ids).await?;
2146    }
2147    Ok(touched)
2148}
2149
2150/// How many times a repair batch that lost a lock race is retried before the
2151/// error surfaces. Each batch is its own transaction, so a retry repeats no
2152/// committed work.
2153const LOCK_CONTENTION_RETRIES: u32 = 5;
2154
2155/// [`recompute_batch`], retried when the database aborted the batch to break a
2156/// lock cycle or a serialisation conflict with a concurrent writer.
2157async fn recompute_batch_retrying(
2158    conn: &mut RuntimeConnection,
2159    view: &SqlView,
2160    ids: &[i64],
2161) -> AutumnResult<usize> {
2162    let mut attempt = 0;
2163    loop {
2164        match recompute_batch(conn, view, ids).await {
2165            Err(error) if attempt < LOCK_CONTENTION_RETRIES && is_lock_contention(&error) => {
2166                attempt += 1;
2167                tracing::warn!(
2168                    parent_table = view.parent_table,
2169                    column = view.counter_column,
2170                    attempt,
2171                    error = %error,
2172                    "repair batch lost a lock race; retrying"
2173                );
2174                tokio::time::sleep(std::time::Duration::from_millis(25 * u64::from(attempt))).await;
2175            }
2176            outcome => return outcome,
2177        }
2178    }
2179}
2180
2181/// Whether `error` is the database aborting one transaction so that another
2182/// can proceed: Postgres's deadlock detector (`40P01`) or a serialisation
2183/// failure (`40001`), and `SQLite`'s busy timeout.
2184pub fn is_lock_contention(error: &AutumnError) -> bool {
2185    let message = error.to_string();
2186    message.contains("deadlock detected")
2187        || message.contains("could not serialize access")
2188        || message.contains("database is locked")
2189}
2190
2191/// Recompute counters from the source of truth.
2192///
2193/// With `parent_id = None` every parent row is rebuilt; with `Some(id)` only
2194/// that parent is touched. Idempotent by construction — the column is *assigned*
2195/// a `COUNT(*)`, never adjusted — so it is safe to run repeatedly, and it is the
2196/// supported way to adopt a counter column on an existing table (AC6).
2197///
2198/// Safe to run against live traffic: each batch locks the parents it is about to
2199/// rebuild before counting, so it can neither read a half-applied write nor
2200/// clobber one (see [`recompute_batch`]).
2201///
2202/// Returns the number of parent rows **actually repaired**, summed across every
2203/// spec — a sweep over a table with no drift returns 0 and writes nothing.
2204///
2205/// # Errors
2206///
2207/// Propagates any database error from the `SELECT`s or `UPDATE`s.
2208#[doc(hidden)]
2209pub async fn counter_cache_recompute<M: 'static>(
2210    conn: &mut RuntimeConnection,
2211    specs: &[CounterCacheSpec<M>],
2212    parent_id: Option<i64>,
2213) -> AutumnResult<usize> {
2214    let mut touched = 0usize;
2215    for index in specs_in_lock_order(specs) {
2216        touched += recompute_view(conn, &view(&specs[index]), parent_id).await?;
2217    }
2218    Ok(touched)
2219}
2220
2221#[derive(diesel::QueryableByName)]
2222struct ChildFkRow {
2223    #[diesel(sql_type = BigInt)]
2224    child_id: i64,
2225    #[diesel(sql_type = Nullable<BigInt>)]
2226    fk_value: Option<i64>,
2227}
2228
2229/// [`ChildFkRow`] plus the row's contribution, for a derivation's bulk capture.
2230#[derive(diesel::QueryableByName)]
2231struct ChildFkContribRow {
2232    #[diesel(sql_type = BigInt)]
2233    child_id: i64,
2234    #[diesel(sql_type = Nullable<BigInt>)]
2235    fk_value: Option<i64>,
2236    #[diesel(sql_type = BigInt)]
2237    contrib_value: i64,
2238}
2239
2240/// A single `i64` primary key, aliased `id`.
2241///
2242/// Shared by the two statements whose result is incidental: the recompute
2243/// sweep's page of parent ids, and the bulk-delete child lock (whose rows are
2244/// discarded — the point is the lock it takes).
2245#[derive(diesel::QueryableByName)]
2246struct IdRow {
2247    #[diesel(sql_type = BigInt)]
2248    id: i64,
2249}
2250
2251#[derive(diesel::QueryableByName)]
2252struct FkRow {
2253    #[diesel(sql_type = Nullable<BigInt>)]
2254    fk_value: Option<i64>,
2255}
2256
2257/// [`FkRow`] plus the row's contribution, for a derivation's single capture.
2258#[derive(diesel::QueryableByName)]
2259struct FkContribRow {
2260    #[diesel(sql_type = Nullable<BigInt>)]
2261    fk_value: Option<i64>,
2262    #[diesel(sql_type = BigInt)]
2263    contrib_value: i64,
2264}
2265
2266/// [`FkContribRow`] plus the row's tenant as text, for a leg that reads it.
2267#[derive(diesel::QueryableByName)]
2268struct FkContribTenantRow {
2269    #[diesel(sql_type = Nullable<BigInt>)]
2270    fk_value: Option<i64>,
2271    #[diesel(sql_type = BigInt)]
2272    contrib_value: i64,
2273    #[diesel(sql_type = Nullable<Text>)]
2274    tenant_text: Option<String>,
2275}
2276
2277impl FkContribTenantRow {
2278    fn captured(self) -> CapturedContribution {
2279        self.fk_value.map(|parent| Captured {
2280            parent,
2281            contrib: self.contrib_value,
2282            tenant: self.tenant_text,
2283        })
2284    }
2285}
2286
2287/// [`FkContribTenantRow`] keyed by child, for the bulk capture.
2288#[derive(diesel::QueryableByName)]
2289struct ChildFkContribTenantRow {
2290    #[diesel(sql_type = BigInt)]
2291    child_id: i64,
2292    #[diesel(sql_type = Nullable<BigInt>)]
2293    fk_value: Option<i64>,
2294    #[diesel(sql_type = BigInt)]
2295    contrib_value: i64,
2296    #[diesel(sql_type = Nullable<Text>)]
2297    tenant_text: Option<String>,
2298}
2299
2300impl ChildFkContribTenantRow {
2301    fn captured(self) -> CapturedContribution {
2302        self.fk_value.map(|parent| Captured {
2303            parent,
2304            contrib: self.contrib_value,
2305            tenant: self.tenant_text,
2306        })
2307    }
2308}
2309
2310/// `Result::optional`, spelled locally so this module does not have to pull
2311/// diesel's `OptionalExtension` into every call site's scope.
2312trait OptionalRow<T> {
2313    fn optional_row(self) -> AutumnResult<Option<T>>;
2314}
2315
2316impl<T> OptionalRow<T> for Result<T, diesel::result::Error> {
2317    fn optional_row(self) -> AutumnResult<Option<T>> {
2318        match self {
2319            Ok(value) => Ok(Some(value)),
2320            Err(diesel::result::Error::NotFound) => Ok(None),
2321            Err(e) => Err(AutumnError::from(e)),
2322        }
2323    }
2324}
2325
2326#[cfg(test)]
2327mod tests {
2328    use super::*;
2329
2330    struct Dummy;
2331
2332    fn spec(soft: bool) -> CounterCacheSpec<Dummy> {
2333        CounterCacheSpec {
2334            child_table: "comments",
2335            child_pk: "id",
2336            child_soft_delete: soft,
2337            fk_column: "post_id",
2338            parent_table: "posts",
2339            parent_pk: "id",
2340            counter_column: "comment_count",
2341            fk_of: |_| Some(1),
2342            pk_of: |_| 1,
2343            live_of: |_| true,
2344            tenant_column: None,
2345            tenant_of: None,
2346            contrib_of: |_| 1,
2347            contrib_sql: "1",
2348            filter_sql: "",
2349            derivation: None,
2350        }
2351    }
2352
2353    /// A filtered `sum(score)` derivation over the same tables.
2354    fn sum_spec() -> CounterCacheSpec<Dummy> {
2355        CounterCacheSpec {
2356            counter_column: "visible_score",
2357            contrib_sql: "{c}.\"score\"",
2358            filter_sql: " AND ({c}.\"published\" = TRUE)",
2359            ..spec(false)
2360        }
2361    }
2362
2363    #[test]
2364    fn plain_identifiers_are_accepted_and_sql_fragments_are_not() {
2365        assert!(is_plain_identifier("comment_count"));
2366        assert!(is_plain_identifier("_x9"));
2367        assert!(!is_plain_identifier(""));
2368        assert!(!is_plain_identifier("9lives"));
2369        assert!(!is_plain_identifier("comment_count; DROP TABLE posts"));
2370        assert!(!is_plain_identifier("comment count"));
2371        assert!(!is_plain_identifier("\"comment_count\""));
2372    }
2373
2374    #[test]
2375    fn the_live_predicate_is_emitted_only_for_a_soft_deleting_child() {
2376        assert_eq!(live_predicate(&view(&spec(false)), true), "");
2377        assert_eq!(
2378            live_predicate(&view(&spec(true)), true),
2379            format!(" AND {CHILD_ALIAS}.\"deleted_at\" IS NULL")
2380        );
2381        assert_eq!(
2382            live_predicate(&view(&spec(true)), false),
2383            format!(" AND {CHILD_ALIAS}.\"deleted_at\" IS NOT NULL")
2384        );
2385    }
2386
2387    #[test]
2388    fn a_bulk_decrement_locks_its_children_in_ascending_id_order() {
2389        // Deterministic order, and children before parents — the same direction
2390        // the single-row and update paths take. Inverting it here would let a
2391        // bulk delete deadlock against an `update` that re-parents one of the
2392        // same children.
2393        let sql = child_lock_sql(&view(&spec(false)), "3,1,2");
2394        assert!(
2395            sql.starts_with("SELECT \"id\" AS id FROM \"comments\""),
2396            "{sql}"
2397        );
2398        assert!(sql.contains("WHERE \"id\" IN (3,1,2)"), "{sql}");
2399        assert!(
2400            sql.contains(&format!("ORDER BY \"id\"{FOR_UPDATE}")),
2401            "the lock must be taken in a deterministic order: {sql}"
2402        );
2403    }
2404
2405    #[test]
2406    fn a_recompute_batch_is_scoped_to_the_ids_it_locked() {
2407        // Restricting the `UPDATE` to the batch is what makes the locking
2408        // `SELECT` that precedes it meaningful: a sweep-wide `UPDATE` would
2409        // touch parents this transaction never locked.
2410        let sql = recompute_update_sql(&view(&spec(false)), "1,2,3");
2411        assert!(sql.contains("\"posts\".\"id\" IN (1,2,3)"), "{sql}");
2412        assert!(
2413            sql.contains(&format!("\"posts\".\"comment_count\" {IS_DISTINCT_FROM}")),
2414            "a healthy parent must still be left unwritten: {sql}"
2415        );
2416    }
2417
2418    /// Two legs whose parents sit in tables named in the *opposite* order to
2419    /// their declaration — the shape a second child model would produce if it
2420    /// declared the same two parents the other way round.
2421    fn two_legs() -> Vec<CounterCacheSpec<Dummy>> {
2422        let mut users = spec(false);
2423        users.parent_table = "users";
2424        users.counter_column = "sent_count";
2425        let mut posts = spec(false);
2426        posts.parent_table = "posts";
2427        posts.counter_column = "comment_count";
2428        // Declared users-first; `posts` sorts first.
2429        vec![users, posts]
2430    }
2431
2432    #[test]
2433    fn deltas_go_out_in_a_globally_stable_lock_order() {
2434        // The order has to come from the schema, not from either model's
2435        // declaration order: two child models declaring the same two parents in
2436        // opposite orders would otherwise take the two row locks in opposite
2437        // orders and deadlock, and the generated transactions do not retry.
2438        let specs = two_legs();
2439        let ordered = fold_and_order(
2440            &specs,
2441            vec![
2442                (0, 5, 1, TenantScope::SameTenantAsChild(100)),
2443                (1, 9, 1, TenantScope::SameTenantAsChild(100)),
2444                (1, 2, 1, TenantScope::SameTenantAsChild(100)),
2445            ],
2446        );
2447        let keys: Vec<(&str, i64)> = ordered
2448            .iter()
2449            .map(|(i, parent_id, _, _)| (specs[*i].parent_table, *parent_id))
2450            .collect();
2451        assert_eq!(keys, vec![("posts", 2), ("posts", 9), ("users", 5)]);
2452
2453        // …and the same for the paths that resolve the parent inside the
2454        // statement, which can only order by table.
2455        assert_eq!(specs_in_lock_order(&specs), vec![1, 0]);
2456    }
2457
2458    #[test]
2459    fn contributions_to_one_parent_fold_into_a_single_statement() {
2460        let specs = two_legs();
2461        // Re-parenting away from and back to the same parent nets out; the two
2462        // remaining moves stay one statement each.
2463        let ordered = fold_and_order(
2464            &specs,
2465            vec![
2466                (1, 2, -1, TenantScope::SameTenantAsChild(10)),
2467                (1, 2, 1, TenantScope::SameTenantAsChild(11)),
2468                (1, 7, 1, TenantScope::SameTenantAsChild(12)),
2469                (1, 7, 1, TenantScope::SameTenantAsChild(13)),
2470                (0, 4, -1, TenantScope::SameTenantAsChild(14)),
2471            ],
2472        );
2473        assert_eq!(
2474            ordered,
2475            vec![
2476                (1, 7, 2, TenantScope::SameTenantAsChild(12)),
2477                (0, 4, -1, TenantScope::SameTenantAsChild(14))
2478            ]
2479        );
2480    }
2481
2482    #[test]
2483    fn a_net_total_no_single_delta_can_carry_goes_out_in_same_sign_pieces() {
2484        // Two children each weighing `i64::MAX` onto one parent: the sum is
2485        // not an `i64`, so a wrapping fold would apply `-2`. The net total is
2486        // carried by two deltas of the same sign instead.
2487        let specs = two_legs();
2488        let ordered = fold_and_order(
2489            &specs,
2490            vec![
2491                (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2492                (1, 2, i64::MAX, TenantScope::SameTenantAsChild(11)),
2493            ],
2494        );
2495        assert_eq!(
2496            ordered,
2497            vec![
2498                (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2499                (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10))
2500            ]
2501        );
2502        // …and a total one past `i64::MAX` is `MAX` then `1`, never a wrapped
2503        // negative: the sum of every delta issued is the sum requested.
2504        let ordered = fold_and_order(
2505            &specs,
2506            vec![
2507                (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2508                (1, 2, -5, TenantScope::SameTenantAsChild(11)),
2509                (1, 2, 5, TenantScope::SameTenantAsChild(12)),
2510                (1, 2, 1, TenantScope::SameTenantAsChild(13)),
2511            ],
2512        );
2513        assert_eq!(
2514            ordered,
2515            vec![
2516                (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2517                (1, 2, 1, TenantScope::SameTenantAsChild(10))
2518            ]
2519        );
2520    }
2521
2522    #[test]
2523    fn weights_that_cancel_across_an_overflowing_intermediate_issue_nothing() {
2524        // `[MAX, MAX, -MAX, -MAX]` nets to zero. Segmenting at the overflow
2525        // point would have issued `+MAX` then `-MAX`, and a parent sitting at
2526        // `1` would have aborted the whole mutation on the first statement
2527        // even though its final value is exactly where it started.
2528        let specs = two_legs();
2529        let ordered = fold_and_order(
2530            &specs,
2531            vec![
2532                (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2533                (1, 2, i64::MAX, TenantScope::SameTenantAsChild(11)),
2534                (1, 2, -i64::MAX, TenantScope::SameTenantAsChild(12)),
2535                (1, 2, -i64::MAX, TenantScope::SameTenantAsChild(13)),
2536            ],
2537        );
2538        assert!(ordered.is_empty(), "{ordered:?}");
2539        // The same shape netting to a representable value is one delta.
2540        let ordered = fold_and_order(
2541            &specs,
2542            vec![
2543                (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2544                (1, 2, i64::MAX, TenantScope::SameTenantAsChild(11)),
2545                (1, 2, -i64::MAX, TenantScope::SameTenantAsChild(12)),
2546                (1, 2, -7, TenantScope::SameTenantAsChild(13)),
2547            ],
2548        );
2549        assert_eq!(
2550            ordered,
2551            vec![(1, 2, i64::MAX - 7, TenantScope::SameTenantAsChild(10))]
2552        );
2553    }
2554
2555    #[test]
2556    fn a_weight_edit_whose_difference_does_not_fit_goes_out_as_two_deltas() {
2557        // `i64::MIN` edited to `i64::MAX` on the same parent: the mathematical
2558        // difference is 2^64 - 1, which is no `i64`, so the old weight comes
2559        // off and the new one goes on as separate statements. Folding keeps
2560        // them apart for the same reason.
2561        let mut out = Vec::new();
2562        push_diff(
2563            0,
2564            Some(captured(7, i64::MIN)),
2565            Some(captured(7, i64::MAX)),
2566            1,
2567            &mut out,
2568        );
2569        assert_eq!(
2570            out,
2571            vec![
2572                (0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
2573                (0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
2574                (0, 7, i64::MAX, TenantScope::SameTenantAsChild(1))
2575            ]
2576        );
2577        // Folded, the three pieces net to 2^64 - 1 and go out as same-sign
2578        // deltas: the parent climbs monotonically to its final value.
2579        let specs = two_legs();
2580        assert_eq!(
2581            fold_and_order(&specs, out.clone()),
2582            vec![
2583                (0, 7, i64::MAX, TenantScope::SameTenantAsChild(1)),
2584                (0, 7, i64::MAX, TenantScope::SameTenantAsChild(1)),
2585                (0, 7, 1, TenantScope::SameTenantAsChild(1))
2586            ]
2587        );
2588
2589        // An ordinary edit is still one delta for the difference.
2590        let mut out = Vec::new();
2591        push_diff(0, Some(captured(7, 3)), Some(captured(7, 10)), 1, &mut out);
2592        assert_eq!(out, vec![(0, 7, 7, TenantScope::SameTenantAsChild(1))]);
2593    }
2594
2595    #[test]
2596    fn removing_the_one_unnegatable_weight_splits_it_in_two() {
2597        // `-i64::MIN` does not exist. Deleting (or re-parenting) a child that
2598        // weighs exactly that much takes it off the old parent as two halves
2599        // whose sum is the whole, rather than wrapping to `i64::MIN` again and
2600        // moving the parent the wrong way by 2^63.
2601        let mut out = Vec::new();
2602        push_diff(0, Some(captured(7, i64::MIN)), None, 1, &mut out);
2603        assert_eq!(
2604            out,
2605            vec![
2606                (0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
2607                (0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1))
2608            ]
2609        );
2610        assert_eq!(
2611            out.iter()
2612                .map(|(_, _, delta, _)| i128::from(*delta))
2613                .sum::<i128>(),
2614            -i128::from(i64::MIN)
2615        );
2616
2617        let mut out = Vec::new();
2618        push_diff(
2619            0,
2620            Some(captured(7, i64::MIN)),
2621            Some(captured(8, 4)),
2622            1,
2623            &mut out,
2624        );
2625        assert_eq!(
2626            out,
2627            vec![
2628                (0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
2629                (0, 7, i64::MAX / 2 + 1, TenantScope::SameTenantAsChild(1)),
2630                (0, 8, 4, TenantScope::SameTenantAsChild(1))
2631            ]
2632        );
2633    }
2634
2635    #[test]
2636    fn a_tenant_scoped_run_is_ordered_so_no_prefix_leaves_the_span_of_zero_and_total() {
2637        // The leg cannot fold (each delta is guarded by its own child's
2638        // tenant), so the order is what keeps `[MAX, MAX, -MAX]` from passing
2639        // through `2 * MAX` on the way to its representable total.
2640        let mut specs = two_legs();
2641        specs[1].tenant_column = Some("tenant_id");
2642        let ordered = fold_and_order(
2643            &specs,
2644            vec![
2645                (1, 2, i64::MAX, TenantScope::SameTenantAsChild(10)),
2646                (1, 2, i64::MAX, TenantScope::SameTenantAsChild(11)),
2647                (1, 2, -i64::MAX, TenantScope::SameTenantAsChild(12)),
2648            ],
2649        );
2650        let deltas: Vec<i64> = ordered.iter().map(|(_, _, d, _)| *d).collect();
2651        assert_eq!(deltas, vec![i64::MAX, -i64::MAX, i64::MAX]);
2652        // …and symmetrically for a negative total.
2653        let ordered = fold_and_order(
2654            &specs,
2655            vec![
2656                (1, 2, -i64::MAX, TenantScope::SameTenantAsChild(10)),
2657                (1, 2, -i64::MAX, TenantScope::SameTenantAsChild(11)),
2658                (1, 2, i64::MAX, TenantScope::SameTenantAsChild(12)),
2659            ],
2660        );
2661        let deltas: Vec<i64> = ordered.iter().map(|(_, _, d, _)| *d).collect();
2662        assert_eq!(deltas, vec![-i64::MAX, i64::MAX, -i64::MAX]);
2663        // Every child keeps its own witness, and other parents are untouched.
2664        let witnesses: Vec<TenantScope> = ordered.iter().map(|(_, _, _, w)| w.clone()).collect();
2665        assert_eq!(
2666            witnesses,
2667            vec![
2668                TenantScope::SameTenantAsChild(10),
2669                TenantScope::SameTenantAsChild(12),
2670                TenantScope::SameTenantAsChild(11)
2671            ]
2672        );
2673    }
2674
2675    fn captured_in(parent: i64, contrib: i64, tenant: Option<&str>) -> Captured {
2676        Captured {
2677            parent,
2678            contrib,
2679            tenant: tenant.map(str::to_owned),
2680        }
2681    }
2682
2683    #[test]
2684    fn a_tenant_change_takes_the_old_weight_off_under_the_old_tenant() {
2685        // The row now says the new tenant, so a live predicate would look for
2686        // the old parent in the wrong tenant and miss it; the removal is
2687        // scoped by the tenant the capture read. The addition is live.
2688        let live = TenantScope::SameTenantAsChild(1);
2689        let was_a = TenantScope::Captured(Some("a".to_owned()));
2690        let mut out = Vec::new();
2691        push_diff(
2692            0,
2693            Some(captured_in(7, 3, Some("a"))),
2694            Some(captured_in(7, 3, Some("b"))),
2695            1,
2696            &mut out,
2697        );
2698        assert_eq!(
2699            out,
2700            vec![(0, 7, -3, was_a.clone()), (0, 7, 3, live.clone())]
2701        );
2702
2703        // Re-parented and moved at once: the same two statements.
2704        let mut out = Vec::new();
2705        push_diff(
2706            0,
2707            Some(captured_in(7, 3, Some("a"))),
2708            Some(captured_in(8, 3, Some("b"))),
2709            1,
2710            &mut out,
2711        );
2712        assert_eq!(
2713            out,
2714            vec![(0, 7, -3, was_a.clone()), (0, 8, 3, live.clone())]
2715        );
2716
2717        // A NULL old tenant is a captured NULL, not "unknown".
2718        let mut out = Vec::new();
2719        push_diff(
2720            0,
2721            Some(captured_in(7, 1, None)),
2722            Some(captured_in(7, 1, Some("b"))),
2723            1,
2724            &mut out,
2725        );
2726        assert_eq!(
2727            out,
2728            vec![
2729                (0, 7, -1, TenantScope::Captured(None)),
2730                (0, 7, 1, live.clone())
2731            ]
2732        );
2733
2734        // Same tenant: the diff is what it always was, one live delta.
2735        let mut out = Vec::new();
2736        push_diff(
2737            0,
2738            Some(captured_in(7, 3, Some("a"))),
2739            Some(captured_in(7, 10, Some("a"))),
2740            1,
2741            &mut out,
2742        );
2743        assert_eq!(out, vec![(0, 7, 7, live)]);
2744
2745        // Unparented after a tenant change: the removal is still under the old
2746        // tenant, and nothing is added.
2747        let mut out = Vec::new();
2748        push_diff(0, Some(captured_in(7, 3, Some("a"))), None, 1, &mut out);
2749        assert_eq!(out, vec![(0, 7, -3, was_a)]);
2750    }
2751
2752    #[test]
2753    fn only_a_table_with_a_leg_onto_itself_is_serialized_and_only_once() {
2754        // Two ordinary legs: nothing to take.
2755        assert!(self_referential_tables(&two_legs()).is_empty());
2756        // Two legs onto their own table, plus one ordinary: that table once,
2757        // and tables in a stable order across mutations.
2758        let mut replies = spec(false);
2759        replies.parent_table = "comments";
2760        replies.counter_column = "reply_count";
2761        let mut likes = spec(false);
2762        likes.parent_table = "comments";
2763        likes.counter_column = "like_count";
2764        let mut users = spec(false);
2765        users.child_table = "avatars";
2766        users.parent_table = "avatars";
2767        assert_eq!(
2768            self_referential_tables(&[users, replies, spec(false), likes]),
2769            vec!["avatars", "comments"]
2770        );
2771    }
2772
2773    #[test]
2774    fn a_captured_tenant_scopes_the_parent_by_its_text() {
2775        let mut tenanted = spec(false);
2776        tenanted.tenant_column = Some("tenant_id");
2777        assert_eq!(
2778            tenant_predicate_captured(&view(&tenanted)),
2779            format!(" AND CAST(\"posts\".\"tenant_id\" AS TEXT) {IS_NOT_DISTINCT_FROM} {PH3}")
2780        );
2781        // Without a tenant column there is no predicate and no bind.
2782        assert_eq!(tenant_predicate_captured(&view(&spec(false))), "");
2783    }
2784
2785    #[test]
2786    fn a_tenant_scoped_leg_keeps_one_statement_per_child() {
2787        // Folding behind a single arbitrary witness would either sweep
2788        // cross-tenant children into the delta or drop legitimate ones, so a
2789        // tenant-scoped spec stays unfolded — still in the global order.
2790        let mut specs = two_legs();
2791        specs[1].tenant_column = Some("tenant_id");
2792        let ordered = fold_and_order(
2793            &specs,
2794            vec![
2795                (1, 3, 1, TenantScope::SameTenantAsChild(20)),
2796                (1, 3, 1, TenantScope::SameTenantAsChild(21)),
2797            ],
2798        );
2799        assert_eq!(
2800            ordered,
2801            vec![
2802                (1, 3, 1, TenantScope::SameTenantAsChild(20)),
2803                (1, 3, 1, TenantScope::SameTenantAsChild(21))
2804            ]
2805        );
2806    }
2807
2808    #[test]
2809    fn a_tenant_discriminator_is_matched_null_safely() {
2810        // A nullable discriminator with NULL on both sides is the same tenant
2811        // (namely none), but `=` yields NULL there, so the maintenance would
2812        // silently do nothing — a counter that quietly stops moving is worse
2813        // than one that errors.
2814        let mut tenanted = spec(false);
2815        tenanted.tenant_column = Some("tenant_id");
2816
2817        let joined = tenant_predicate_joined(&view(&tenanted));
2818        assert!(joined.contains(IS_NOT_DISTINCT_FROM), "{joined}");
2819        assert!(
2820            !joined.contains("\"tenant_id\" = "),
2821            "plain equality is not NULL-safe: {joined}"
2822        );
2823
2824        // The parent-keyed form still requires the child row to exist, so a
2825        // missing child remains a no-op rather than matching every untenanted
2826        // parent the way a scalar sub-select would.
2827        let keyed = tenant_predicate(&view(&tenanted), 7);
2828        assert!(keyed.contains("EXISTS"), "{keyed}");
2829        assert!(keyed.contains(IS_NOT_DISTINCT_FROM), "{keyed}");
2830        assert!(keyed.contains("\"id\" = 7"), "{keyed}");
2831
2832        // Still nothing at all for an association that declares no tenant.
2833        assert_eq!(tenant_predicate_joined(&view(&spec(false))), "");
2834        assert_eq!(tenant_predicate(&view(&spec(false)), 7), "");
2835    }
2836
2837    #[test]
2838    fn a_counter_column_named_after_a_sql_keyword_still_produces_valid_sql() {
2839        // `counter_cache = "order"` is a legal identifier and a legal column
2840        // name, but `SET order = order + $1` is a syntax error on both backends.
2841        // Quoting every interpolated identifier is what keeps the choice of
2842        // column name from being able to break the generated statements.
2843        let mut keyword = spec(false);
2844        keyword.counter_column = "order";
2845        keyword.parent_table = "group";
2846        let sql = recompute_update_sql(&view(&keyword), "1");
2847        assert!(
2848            sql.starts_with("UPDATE \"group\" SET \"order\" = "),
2849            "{sql}"
2850        );
2851        assert!(sql.contains("\"group\".\"order\" "), "{sql}");
2852        assert!(
2853            !sql.contains(" order "),
2854            "no bare keyword may survive: {sql}"
2855        );
2856    }
2857
2858    #[test]
2859    fn a_plain_counter_cache_keeps_its_pre_derivation_sql() {
2860        // The whole design rests on this: derivations reuse the counter-cache
2861        // paths, so a counter cache's own SQL must be untouched by them.
2862        let plain = view(&spec(false));
2863        assert!(plain.is_plain());
2864        assert!(plain.counts_rows());
2865        assert_eq!(filter_predicate(&plain, CHILD_ALIAS), "");
2866        assert_eq!(aggregate_expr(&plain, CHILD_ALIAS), "COUNT(*)");
2867        assert!(
2868            !recompute_update_sql(&plain, "1").contains("COALESCE"),
2869            "a plain counter cache must not grow a COALESCE"
2870        );
2871        assert_eq!(
2872            accumulator(&plain, "\"comment_count\""),
2873            "\"comment_count\"",
2874            "a plain counter cache reads its column bare"
2875        );
2876        assert!(
2877            !bulk_decrement_sql(&plain, "7").contains("COALESCE"),
2878            "a plain bulk decrement must not grow a COALESCE"
2879        );
2880    }
2881
2882    #[test]
2883    fn a_derivation_delta_never_writes_null_into_the_maintained_column() {
2884        // A derivation is adopted on an existing table, so the column can still
2885        // be NULL. `NULL +/- expr` is NULL, and a NULL maintained value cannot
2886        // be told apart from drift, so every adjusting statement coalesces.
2887        let mut filtered = spec(false);
2888        filtered.filter_sql = " AND ({c}.\"published\" = TRUE)";
2889        let filtered = view(&filtered);
2890        assert_eq!(
2891            accumulator(&filtered, "\"comment_count\""),
2892            "COALESCE(\"comment_count\", 0)"
2893        );
2894        assert!(
2895            bulk_decrement_sql(&filtered, "7")
2896                .contains("SET \"comment_count\" = COALESCE(\"comment_count\", 0) -"),
2897            "{}",
2898            bulk_decrement_sql(&filtered, "7")
2899        );
2900        let sum = view(&sum_spec());
2901        assert!(
2902            weighted_delta_by_child_id_sql(&sum, 1, false, "", "")
2903                .contains("SET \"visible_score\" = COALESCE(\"visible_score\", 0) +"),
2904            "the by-id weighted delta coalesces too"
2905        );
2906    }
2907
2908    #[test]
2909    fn a_lowered_contribution_is_either_one_or_a_single_child_column() {
2910        // The run-time backstop for `contrib_sql`: a structural check, so a
2911        // legal quoted literal is never mistaken for smuggled SQL.
2912        assert!(is_lowered_contribution("1"));
2913        assert!(is_lowered_contribution("{c}.\"score\""));
2914        assert!(!is_lowered_contribution("{c}.\"score\" + 1"));
2915        assert!(!is_lowered_contribution("(SELECT 1)"));
2916        assert!(!is_lowered_contribution(""));
2917        assert!(!is_lowered_contribution("{c}.\"a\"; DROP TABLE posts --\""));
2918    }
2919
2920    #[test]
2921    fn a_filter_may_carry_a_comment_marker_inside_a_quoted_literal() {
2922        // The previous substring scan panicked on this in debug builds: `--`
2923        // and `;` are inert inside a quoted literal, and the macro escapes the
2924        // literal it lowers, so only the identifiers and the contribution are
2925        // checked at run time.
2926        let mut filtered = spec(false);
2927        filtered.filter_sql = " AND ({c}.\"status\" = 'a -- b; c')";
2928        let filtered = view(&filtered);
2929        assert_spec_idents(&filtered);
2930        assert!(
2931            bulk_decrement_sql(&filtered, "7").contains("'a -- b; c'"),
2932            "the literal reaches the statement unaltered"
2933        );
2934    }
2935
2936    #[test]
2937    fn a_filtered_count_carries_its_filter_into_every_statement() {
2938        let mut filtered = spec(false);
2939        filtered.filter_sql = " AND ({c}.\"published\" = TRUE)";
2940        let filtered = view(&filtered);
2941        assert!(
2942            !filtered.is_plain(),
2943            "a filter is not a plain counter cache"
2944        );
2945        assert!(filtered.counts_rows(), "a filtered count still counts rows");
2946
2947        // The deltas: the filter sits in both sub-selects, so a rejected row
2948        // matches no parent and the statement writes nothing.
2949        let delete = bulk_decrement_sql(&filtered, "7,8");
2950        assert_eq!(
2951            delete,
2952            "UPDATE \"posts\" SET \"comment_count\" = \
2953             COALESCE(\"comment_count\", 0) - \
2954             (SELECT COUNT(*) FROM \"comments\" AS __autumn_cc_child \
2955              WHERE __autumn_cc_child.\"post_id\" = \"posts\".\"id\" \
2956                AND __autumn_cc_child.\"id\" IN (7,8) \
2957                AND (__autumn_cc_child.\"published\" = TRUE)) \
2958             WHERE \"posts\".\"id\" IN \
2959               (SELECT __autumn_cc_child.\"post_id\" FROM \"comments\" AS __autumn_cc_child \
2960                WHERE __autumn_cc_child.\"id\" IN (7,8) \
2961                  AND __autumn_cc_child.\"post_id\" IS NOT NULL \
2962                  AND (__autumn_cc_child.\"published\" = TRUE) \
2963                  AND __autumn_cc_child.\"post_id\" = \"posts\".\"id\")"
2964        );
2965
2966        // The capture reads the contribution as a column, so a filter flip on an
2967        // unchanged parent is visible as `0 -> 1`.
2968        assert_eq!(
2969            contrib_case_expr(&filtered, CHILD_ALIAS),
2970            "CAST(CASE WHEN 1 = 1 AND (__autumn_cc_child.\"published\" = TRUE) \
2971             THEN 1 ELSE 0 END AS BIGINT)"
2972        );
2973    }
2974
2975    #[cfg(not(feature = "sqlite"))]
2976    #[test]
2977    fn the_postgres_sum_is_a_plain_coalesced_sum() {
2978        assert_eq!(sum_expr("x"), "COALESCE(SUM(x), 0)");
2979    }
2980
2981    #[test]
2982    fn a_sum_recompute_assigns_the_summed_contribution() {
2983        let sum = view(&sum_spec());
2984        assert!(!sum.counts_rows());
2985        let score_sum = sum_expr("__autumn_cc_child.\"score\"");
2986        assert_eq!(
2987            recompute_update_sql(&sum, "1,2"),
2988            format!(
2989                "UPDATE \"posts\" SET \"visible_score\" = \
2990                 (SELECT {score_sum} \
2991                  FROM \"comments\" AS __autumn_cc_child \
2992                  WHERE __autumn_cc_child.\"post_id\" = \"posts\".\"id\" \
2993                    AND (__autumn_cc_child.\"published\" = TRUE)) \
2994                 WHERE \"posts\".\"id\" IN (1,2) \
2995                   AND \"posts\".\"visible_score\" {IS_DISTINCT_FROM} \
2996                 (SELECT {score_sum} \
2997                  FROM \"comments\" AS __autumn_cc_child \
2998                  WHERE __autumn_cc_child.\"post_id\" = \"posts\".\"id\" \
2999                    AND (__autumn_cc_child.\"published\" = TRUE))"
3000            )
3001        );
3002    }
3003
3004    #[test]
3005    fn a_weighted_by_id_delta_reads_the_weight_from_the_row() {
3006        // The magnitude cannot be bound by the caller: only the database can see
3007        // the row the weight comes from. `COALESCE` keeps a NULL out of the
3008        // maintained column even so.
3009        let sql = weighted_delta_by_child_id_sql(&view(&sum_spec()), 9, false, "", "");
3010        assert!(
3011            sql.starts_with(
3012                "UPDATE \"posts\" SET \"visible_score\" = \
3013                 COALESCE(\"visible_score\", 0) + \
3014                 COALESCE((SELECT __autumn_cc_child.\"score\""
3015            ),
3016            "{sql}"
3017        );
3018        assert_eq!(
3019            sql.matches("__autumn_cc_child.\"id\" = 9").count(),
3020            2,
3021            "{sql}"
3022        );
3023        assert!(
3024            sql.contains(&format!(
3025                "IS NOT NULL AND (__autumn_cc_child.\"published\" = TRUE){FOR_UPDATE})"
3026            )),
3027            "{sql}"
3028        );
3029
3030        let down = weighted_delta_by_child_id_sql(&view(&sum_spec()), 9, true, "", "");
3031        assert!(
3032            down.contains("\"visible_score\" = COALESCE(\"visible_score\", 0) - COALESCE("),
3033            "{down}"
3034        );
3035    }
3036
3037    #[test]
3038    fn a_filter_flip_on_the_same_parent_is_one_delta() {
3039        let mut out = Vec::new();
3040        // Unpublished (0) -> published (1) with the parent unchanged.
3041        push_diff(0, Some(captured(4, 0)), Some(captured(4, 1)), 11, &mut out);
3042        assert_eq!(out, vec![(0, 4, 1, TenantScope::SameTenantAsChild(11))]);
3043
3044        // A rejected row that stays rejected moves nothing, even across a
3045        // reparent: neither side has a weight to move.
3046        out.clear();
3047        push_diff(0, Some(captured(4, 0)), Some(captured(5, 0)), 11, &mut out);
3048        assert!(out.is_empty(), "{out:?}");
3049
3050        // A reparent of a qualifying row moves both ends.
3051        out.clear();
3052        push_diff(0, Some(captured(4, 3)), Some(captured(5, 3)), 11, &mut out);
3053        assert_eq!(
3054            out,
3055            vec![
3056                (0, 4, -3, TenantScope::SameTenantAsChild(11)),
3057                (0, 5, 3, TenantScope::SameTenantAsChild(11))
3058            ]
3059        );
3060
3061        // A weight edit on an unchanged parent is the difference only.
3062        out.clear();
3063        push_diff(0, Some(captured(4, 3)), Some(captured(4, 10)), 11, &mut out);
3064        assert_eq!(out, vec![(0, 4, 7, TenantScope::SameTenantAsChild(11))]);
3065    }
3066
3067    #[test]
3068    fn drift_is_one_aggregate_over_the_parent_table() {
3069        let sql = drift_sql(&view(&sum_spec()), 10_000);
3070        assert!(
3071            sql.starts_with("SELECT COUNT(*) AS count FROM (SELECT 1 AS drifted FROM "),
3072            "{sql}"
3073        );
3074        assert!(sql.contains(IS_DISTINCT_FROM), "{sql}");
3075        // The cap is on the parents examined, not on the matches: a healthy
3076        // table has no match, so an output-side LIMIT would still evaluate the
3077        // correlated aggregate for every parent before giving up.
3078        assert!(
3079            sql.contains(
3080                "FROM (SELECT * FROM \"posts\" ORDER BY \"posts\".\"id\" DESC LIMIT 10000) AS \"posts\" WHERE"
3081            ),
3082            "the window is materialised under the table's own name: {sql}"
3083        );
3084        assert!(sql.ends_with(") AS __autumn_cc_drift"), "{sql}");
3085    }
3086
3087    #[test]
3088    fn a_model_without_an_inherent_shadow_resolves_to_the_empty_blanket() {
3089        // A `const` block, because the whole point of the flag is that it folds
3090        // at compile time — `assert!` on a constant is a clippy error.
3091        const { assert!(!Dummy::HAS_COUNTER_CACHES) };
3092        assert!(Dummy::counter_caches().is_empty());
3093    }
3094
3095    /// The `SQLite` sum aggregate, run against a real database (#2663).
3096    #[cfg(feature = "sqlite")]
3097    mod sqlite_sum {
3098        use super::{CHILD_ALIAS, aggregate_expr, filter_predicate, sum_spec, view};
3099        use diesel::Connection as _;
3100        use proptest::prelude::*;
3101
3102        #[derive(diesel::QueryableByName)]
3103        struct SumRow {
3104            #[diesel(sql_type = diesel::sql_types::BigInt)]
3105            total: i64,
3106        }
3107
3108        /// Sum `rows` (weight, published) with the derivation's aggregate, in
3109        /// row order.
3110        fn sqlite_sum(rows: &[(i64, bool)]) -> Result<i64, diesel::result::Error> {
3111            let mut conn =
3112                diesel::SqliteConnection::establish(":memory:").expect("in-memory database");
3113            diesel::RunQueryDsl::execute(
3114                diesel::sql_query(
3115                    "CREATE TABLE comments (id INTEGER PRIMARY KEY, \
3116                     score BIGINT NOT NULL, published BOOLEAN NOT NULL)",
3117                ),
3118                &mut conn,
3119            )
3120            .expect("DDL");
3121            for (score, published) in rows {
3122                diesel::RunQueryDsl::execute(
3123                    diesel::sql_query("INSERT INTO comments (score, published) VALUES (?, ?)")
3124                        .bind::<diesel::sql_types::BigInt, _>(score)
3125                        .bind::<diesel::sql_types::Bool, _>(published),
3126                    &mut conn,
3127                )
3128                .expect("seed row");
3129            }
3130            let view = view(&sum_spec());
3131            let sql = format!(
3132                "SELECT {} AS total FROM \"comments\" AS {CHILD_ALIAS} WHERE 1 = 1{}",
3133                aggregate_expr(&view, CHILD_ALIAS),
3134                filter_predicate(&view, CHILD_ALIAS)
3135            );
3136            diesel::RunQueryDsl::get_result::<SumRow>(diesel::sql_query(sql), &mut conn)
3137                .map(|row| row.total)
3138        }
3139
3140        /// The exact sum of the published weights, if it fits in `i64`.
3141        fn exact_sum(rows: &[(i64, bool)]) -> Option<i64> {
3142            let total: i128 = rows
3143                .iter()
3144                .filter(|(_, published)| *published)
3145                .map(|(score, _)| i128::from(*score))
3146                .sum();
3147            i64::try_from(total).ok()
3148        }
3149
3150        fn published(scores: &[i64]) -> Vec<(i64, bool)> {
3151            scores.iter().map(|score| (*score, true)).collect()
3152        }
3153
3154        #[test]
3155        fn an_empty_set_sums_to_zero() {
3156            assert_eq!(sqlite_sum(&[]).expect("sum"), 0);
3157            assert_eq!(sqlite_sum(&[(5, false)]).expect("sum"), 0);
3158        }
3159
3160        #[test]
3161        fn a_partial_sum_past_max_does_not_fail_the_scan() {
3162            assert_eq!(
3163                sqlite_sum(&published(&[i64::MAX, 1, -1])).expect("sum"),
3164                i64::MAX
3165            );
3166        }
3167
3168        #[test]
3169        fn a_partial_sum_past_min_does_not_fail_the_scan() {
3170            assert_eq!(
3171                sqlite_sum(&published(&[i64::MIN, -1, 1])).expect("sum"),
3172                i64::MIN
3173            );
3174        }
3175
3176        #[test]
3177        fn a_total_that_does_not_fit_is_an_error() {
3178            for scores in [[i64::MAX, 1], [i64::MIN, -1]] {
3179                let err = sqlite_sum(&published(&scores)).expect_err("no i64 holds the total");
3180                assert!(err.to_string().contains("integer overflow"), "{err}");
3181            }
3182        }
3183
3184        fn weight() -> impl Strategy<Value = i64> {
3185            prop_oneof![
3186                Just(i64::MAX),
3187                Just(i64::MIN),
3188                Just(0),
3189                Just(1),
3190                Just(-1),
3191                Just(1 << 32),
3192                Just(-(1 << 32)),
3193                Just((1 << 32) - 1),
3194                any::<i64>(),
3195            ]
3196        }
3197
3198        /// Seed one post and its `rows` (weight, published), set the post to
3199        /// `start`, then bulk-decrement the rows with ids in `deleted`.
3200        fn sqlite_bulk_decrement(
3201            rows: &[(i64, bool)],
3202            start: i64,
3203            deleted: &[i64],
3204        ) -> Result<i64, diesel::result::Error> {
3205            let mut conn =
3206                diesel::SqliteConnection::establish(":memory:").expect("in-memory database");
3207            for ddl in [
3208                "CREATE TABLE posts (id INTEGER PRIMARY KEY, visible_score BIGINT NOT NULL)",
3209                "CREATE TABLE comments (id INTEGER PRIMARY KEY, post_id BIGINT NOT NULL, \
3210                 score BIGINT NOT NULL, published BOOLEAN NOT NULL)",
3211            ] {
3212                diesel::RunQueryDsl::execute(diesel::sql_query(ddl), &mut conn).expect("DDL");
3213            }
3214            diesel::RunQueryDsl::execute(
3215                diesel::sql_query("INSERT INTO posts (id, visible_score) VALUES (1, ?)")
3216                    .bind::<diesel::sql_types::BigInt, _>(start),
3217                &mut conn,
3218            )
3219            .expect("seed post");
3220            for (score, published) in rows {
3221                diesel::RunQueryDsl::execute(
3222                    diesel::sql_query(
3223                        "INSERT INTO comments (post_id, score, published) VALUES (1, ?, ?)",
3224                    )
3225                    .bind::<diesel::sql_types::BigInt, _>(score)
3226                    .bind::<diesel::sql_types::Bool, _>(published),
3227                    &mut conn,
3228                )
3229                .expect("seed row");
3230            }
3231            let sql = super::bulk_decrement_sql(&view(&sum_spec()), &super::id_list(deleted));
3232            diesel::RunQueryDsl::execute(diesel::sql_query(sql), &mut conn)?;
3233            diesel::RunQueryDsl::get_result::<SumRow>(
3234                diesel::sql_query("SELECT visible_score AS total FROM posts WHERE id = 1"),
3235                &mut conn,
3236            )
3237            .map(|row| row.total)
3238        }
3239
3240        #[test]
3241        fn a_bulk_decrement_whose_batch_sum_leaves_i64_still_lands() {
3242            // MAX + MAX + MIN = MAX - 1. Deleting both MAX rows leaves MIN.
3243            let rows = published(&[i64::MAX, i64::MAX, i64::MIN]);
3244            assert_eq!(
3245                sqlite_bulk_decrement(&rows, i64::MAX - 1, &[1, 2]).expect("decrement"),
3246                i64::MIN
3247            );
3248        }
3249
3250        #[test]
3251        fn a_bulk_decrement_whose_result_leaves_i64_is_an_error() {
3252            let err = sqlite_bulk_decrement(&published(&[1]), i64::MIN, &[1])
3253                .expect_err("MIN - 1 has no i64 value");
3254            assert!(err.to_string().contains("integer overflow"), "{err}");
3255        }
3256
3257        proptest! {
3258            #[test]
3259            fn the_sum_is_exact_or_an_overflow_error(
3260                rows in prop::collection::vec((weight(), any::<bool>()), 0..8)
3261            ) {
3262                match (exact_sum(&rows), sqlite_sum(&rows)) {
3263                    (Some(want), got) => prop_assert_eq!(got.ok(), Some(want)),
3264                    (None, Ok(got)) => prop_assert!(false, "{rows:?} gave {got}"),
3265                    (None, Err(err)) => {
3266                        prop_assert!(err.to_string().contains("integer overflow"), "{err}");
3267                    }
3268                }
3269            }
3270
3271            #[test]
3272            fn the_bulk_decrement_is_exact_or_an_overflow_error(
3273                rows in prop::collection::vec((weight(), any::<bool>(), any::<bool>()), 0..8)
3274            ) {
3275                let seeded: Vec<(i64, bool)> = rows.iter().map(|(w, p, _)| (*w, *p)).collect();
3276                // The parent holds the exact total, as the delta paths leave it.
3277                let Some(start) = exact_sum(&seeded) else {
3278                    return Ok(());
3279                };
3280                let deleted: Vec<i64> = (1_i64..)
3281                    .zip(&rows)
3282                    .filter(|(_, (_, _, delete))| *delete)
3283                    .map(|(id, _)| id)
3284                    .collect();
3285                let kept: Vec<(i64, bool)> = rows
3286                    .iter()
3287                    .filter(|(_, _, delete)| !*delete)
3288                    .map(|(w, p, _)| (*w, *p))
3289                    .collect();
3290                let want = exact_sum(&kept);
3291                let got = sqlite_bulk_decrement(&seeded, start, &deleted);
3292                match (want, got) {
3293                    (Some(want), got) => prop_assert_eq!(got.ok(), Some(want)),
3294                    (None, Ok(got)) => prop_assert!(false, "{rows:?} gave {got}"),
3295                    (None, Err(err)) => {
3296                        prop_assert!(err.to_string().contains("integer overflow"), "{err}");
3297                    }
3298                }
3299            }
3300        }
3301    }
3302}