CHANGELOG.md

@@ -5,6 +5,7 @@ All notable changes are documented here. This project follows Keep a Changelog a
 ## [Unreleased]
 
 ### Added
+- QQQ attribution and N+1 detection: a core statement execution listener (QQQ 4.1.0-RC.13) aggregates database time per route, scheduled job, process, table and statement shape in bounded memory, records repeated statement shapes per unit of work, and each pod flushes one batched insert per table every `attribution_flush_seconds`. Hosts without QQQ's application launcher call `DbPerfRuntime.of(instance).startAttribution()` after startup.
 
 - Secured database-performance tables, PostgreSQL statistics snapshots, daily rollups and configurable retention.
 - Audited PLAN/GENERIC and read-only ANALYZE query viewer with SQL validation, timeout, rollback and transient parameter handling.

README.md

@@ -4,7 +4,7 @@ Database performance for QQQ apps: PostgreSQL statement deltas, table and index
 
 ## Requirements
 
-- Java 21 and QQQ **4.1.0-RC.11+**.
+- Java 21 and QQQ **4.1.0-RC.13+** (the statement execution listener SPI used for QQQ attribution).
 - PostgreSQL **13+**; **16+** for parameterized generic plans, with the pgJDBC driver (built against 42.7.14).
 - `pg_stat_statements` preloaded, created in the monitored database, and readable by the application role.
 - `pg_read_all_stats` granted to that role for complete statistics and other sessions' visibility.

docs/ATTRIBUTION.md

@@ -0,0 +1,58 @@
+# Attribution runtime (phase B2)
+
+The producer registers `AttributionRuntimeService` with the QQQ application
+launcher. It starts one daemon per application instance after initialization,
+independently of scheduled-job leadership. Hosts without `QApplicationLauncher`
+should call `DbPerfRuntime.of(instance).startAttribution()` after migrations and
+instance validation, and `stopAttribution()` before disposing of the instance.
+`attributionEnabled=false` prevents listener registration and background work.
+Metadata production itself starts no threads, so existing metadata tests remain
+isolated.
+
+`AttributionFlusher` can also be constructed directly for deterministic tests:
+`refreshSettings()`, register `getListener()` with `StatementExecutionListeners`,
+and call `flushNow()`. Always close it to unregister. Production `start()` loads
+settings, registers the listener, schedules flushing, and installs a JVM shutdown
+hook. Closing stops scheduling, unregisters, and flushes once. Shutdown only
+persists completed units; hosts should drain application requests first.
+
+Statements never read settings from the database. A successful settings refresh
+publishes an immutable cached copy; a failed refresh keeps the previous copy.
+The initial copy disables recording until settings can be read. The configured
+flush interval takes effect when the next pass is scheduled. Disabled attribution
+still refreshes settings so administrators can re-enable it.
+
+Units are sampled once and buffered until completion, allowing HTTP route names
+to be finalized. A completed unit belongs to the window in which it ends; an
+abandoned unit is merged when that worker sees a different unit. No-unit work is
+sampled per statement and counted as one unit per statement. Sampling transitions
+retain separate aggregates for each effective rate. Values are observed counts,
+not extrapolated estimates. Daily rollups sum those counts across windows and
+pods and take the maximum duration.
+
+Each global map includes its overflow bucket within `attribution_max_keys`.
+One slot is reserved for a universal `(other)` key (all dimensions collapse,
+fingerprint is SHA-256 of `(other)`). Per-unit maps use the same bound with a
+minimum of two slots. Collapsed per-unit shapes cannot establish N+1 evidence
+and are excluded from repeat detection, avoiding false positives. Global repeat
+overflow sums qualifying repetitions but counts each affected unit only once.
+A mixed-rate overflow row has a null sample rate because there is no single
+valid rate for it. The SQL normalization cache holds at most 5,000 raw SQL keys
+in memory and clears on overflow; only normalized SQL is persisted, capped at
+4,000 characters.
+
+Window swaps and counter merges synchronize only in memory. Database work runs
+under the system user on the flusher thread with attribution suppressed. Each
+nonempty table receives one `InsertAction`, even above 1,000 rows; QQQ may page
+the underlying SQL. Both tables are written in one transaction. Failed windows
+are dropped, with rate-limited fixed-text logging, and never queued for retry.
+The repeat-pattern migration adds `sample_rate` without changing prior
+changesets. Existing retention already includes repeat patterns.
+
+Build using the locally installed statement-listener API:
+
+```powershell
+.\mvnw.cmd -o verify '-Dqqq.snapshot.version=4.1.0-dbperf-SNAPSHOT'
+```
+
+The normal POM default is `4.1.0-RC.13`.

pom.xml

@@ -241,14 +241,14 @@
    <profiles>
       <!-- Built against qqq 4.1 (QBitMetaDataProducer). Until qbit-build-parent  -->
       <!-- imports the released 4.1 BOM, this profile is active by default and     -->
-      <!-- imports qqq-bom-pom 4.1.0-RC.11 ahead of the parent's 4.0.0 BOM.         -->
+      <!-- imports qqq-bom-pom 4.1.0-RC.13 ahead of the parent's 4.0.0 BOM.         -->
       <profile>
          <id>qqq-snapshot</id>
          <activation>
             <activeByDefault>true</activeByDefault>
          </activation>
          <properties>
-            <qqq.snapshot.version>4.1.0-RC.11</qqq.snapshot.version>
+            <qqq.snapshot.version>4.1.0-RC.13</qqq.snapshot.version>
          </properties>
          <dependencyManagement>
             <dependencies>

src/main/java/com/kingsrook/qbits/dbperformance/DbPerfQBitProducer.java

@@ -25,6 +25,7 @@ import java.util.ArrayList;
 import java.util.List;
 import com.kingsrook.qbits.dbperformance.alerts.SnapshotEvaluator;
 import com.kingsrook.qbits.dbperformance.api.DbPerformanceUiApi;
+import com.kingsrook.qbits.dbperformance.attribution.AttributionRuntimeService;
 import com.kingsrook.qbits.dbperformance.explain.QueryPlanProcess;
 import com.kingsrook.qbits.dbperformance.health.DbPerfHealthChecks;
 import com.kingsrook.qbits.dbperformance.metadata.DbPerfHelp;
@@ -34,6 +35,7 @@ import com.kingsrook.qbits.dbperformance.widgets.DbPerfWidgets;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.model.metadata.MetaDataProducerMultiOutput;
 import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
 import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitConfigValidationException;
 import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaDataProducer;
@@ -144,6 +146,7 @@ public class DbPerfQBitProducer implements QBitMetaDataProducer<DbPerfQBitConfig
       instance.addQBit(getQBitMetaData());
       DbPerfMetaData.addTo(instance, qBitConfig);
       new DbPerfRuntime(instance, qBitConfig).addSelfToInstance(instance);
+      instance.withRuntimeService(new QCodeReference(AttributionRuntimeService.class));
       DbPerfProcesses.addTo(instance, qBitConfig);
       SnapshotEvaluator.register(instance, qBitConfig);
       DbPerfHealthChecks.register(instance);

src/main/java/com/kingsrook/qbits/dbperformance/DbPerfRuntime.java

@@ -22,6 +22,7 @@ package com.kingsrook.qbits.dbperformance;
 
 
 import java.util.concurrent.atomic.AtomicReference;
+import com.kingsrook.qbits.dbperformance.attribution.AttributionFlusher;
 import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
 import com.kingsrook.qqq.backend.core.context.CapturedContext;
 import com.kingsrook.qqq.backend.core.context.QContext;
@@ -39,6 +40,7 @@ public class DbPerfRuntime implements QSupplementalInstanceMetaData
    public static final String NAME = "dbPerformance";
    private final QInstance instance;
    private final DbPerfQBitConfig config;
+   private AttributionFlusher attributionFlusher;
 
 
    /*******************************************************************************
@@ -140,6 +142,32 @@ public class DbPerfRuntime implements QSupplementalInstanceMetaData
    }
 
 
+   /*******************************************************************************
+    ** Called by the launcher, or explicitly by hosts without a launcher.
+    *******************************************************************************/
+   public synchronized void startAttribution()
+   {
+      if(attributionFlusher == null && Boolean.TRUE.equals(config.getAttributionEnabled()))
+      {
+         attributionFlusher = new AttributionFlusher(this);
+         attributionFlusher.start();
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Stop the per-instance listener and its daemon; safe to call repeatedly.
+    *******************************************************************************/
+   public synchronized void stopAttribution()
+   {
+      if(attributionFlusher != null)
+      {
+         attributionFlusher.close();
+         attributionFlusher = null;
+      }
+   }
+
+
    /*******************************************************************************
     ** Checked service operation.
     *******************************************************************************/

src/main/java/com/kingsrook/qbits/dbperformance/attribution/AttributionFlusher.java

@@ -0,0 +1,270 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dbperformance.attribution;
+
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import com.kingsrook.qbits.dbperformance.DbPerfRuntime;
+import com.kingsrook.qbits.dbperformance.collect.DbPerfSettingsService;
+import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.instrumentation.StatementExecutionListeners;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+
+
+/*******************************************************************************
+ ** One daemon per runtime. Windows are detached before any database operation;
+ ** failed windows are dropped and later windows continue without retry queues.
+ *******************************************************************************/
+public class AttributionFlusher implements AutoCloseable
+{
+   private static final QLogger LOG = QLogger.getLogger(AttributionFlusher.class);
+   private final DbPerfRuntime runtime;
+   private final AttributionListener listener;
+   private ScheduledExecutorService executor;
+   private Thread shutdownHook;
+   private boolean closed;
+   private long failures;
+
+
+   /*******************************************************************************
+    ** Construct without starting threads, for hosts and deterministic tests.
+    *******************************************************************************/
+   public AttributionFlusher(DbPerfRuntime runtime)
+   {
+      this.runtime = runtime;
+      this.listener = new AttributionListener(runtime);
+   }
+
+
+   /*******************************************************************************
+    ** Register once after the instance and schema are ready.
+    *******************************************************************************/
+   public synchronized void start()
+   {
+      if(executor != null || closed || !Boolean.TRUE.equals(runtime.getConfig().getAttributionEnabled()))
+      {
+         return;
+      }
+      refreshSettings();
+      StatementExecutionListeners.register(listener);
+      executor = Executors.newSingleThreadScheduledExecutor(task ->
+      {
+         Thread thread = new Thread(task, "db-perf-attribution");
+         thread.setDaemon(true);
+         return (thread);
+      });
+      shutdownHook = new Thread(this::close, "db-perf-attribution-shutdown");
+      Runtime.getRuntime().addShutdownHook(shutdownHook);
+      schedule();
+   }
+
+
+   /*******************************************************************************
+    ** Reschedule after each pass so administrative interval changes take effect.
+    *******************************************************************************/
+   private synchronized void schedule()
+   {
+      if(!closed)
+      {
+         executor.schedule(() ->
+         {
+            flushNow();
+            schedule();
+         }, listener.settings().flushSeconds(), TimeUnit.SECONDS);
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Exactly one InsertAction per nonempty table, with all rows in the window.
+    ** Serialize flushers only; statement threads never acquire this monitor.
+    *******************************************************************************/
+   public synchronized void flushNow()
+   {
+      boolean previous = AttributionListener.suppress(true);
+      try
+      {
+         Instant end = Instant.now();
+         AttributionListener.Window window = listener.drain(end);
+         List<QRecord> attribution = records(window.attribution(), window.start(), end, false);
+         List<QRecord> repeats = records(window.repeats(), window.start(), end, true);
+         if(!attribution.isEmpty() || !repeats.isEmpty())
+         {
+            runtime.transaction(() ->
+            {
+               insert("db_perf_attribution", attribution);
+               insert("db_perf_repeat_pattern", repeats);
+               return (null);
+            });
+         }
+      }
+      catch(Exception e)
+      {
+         failed();
+      }
+      finally
+      {
+         refreshSettings();
+         AttributionListener.suppress(previous);
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Refresh even after failed writes, including while attribution is disabled.
+    *******************************************************************************/
+   public synchronized void refreshSettings()
+   {
+      boolean previous = AttributionListener.suppress(true);
+      try
+      {
+         listener.refresh(new DbPerfSettingsService(runtime).ensure(runtime.getConfig().getBackendName()));
+      }
+      catch(Exception e)
+      {
+         failed();
+      }
+      finally
+      {
+         AttributionListener.suppress(previous);
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Rate-limited fixed text; database errors may contain sensitive SQL.
+    *******************************************************************************/
+   private void failed()
+   {
+      failures++;
+      if(failures == 1 || failures % 1000 == 0)
+      {
+         LOG.warn("Database Performance attribution write or settings refresh failed; the failed window is dropped");
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Convert nanoseconds to persisted milliseconds without floating point loss.
+    *******************************************************************************/
+   private List<QRecord> records(Map<AttributionListener.Key, AttributionListener.Counters> values, Instant start, Instant end, boolean repeat)
+   {
+      List<QRecord> records = new ArrayList<>(values.size());
+      for(Map.Entry<AttributionListener.Key, AttributionListener.Counters> entry : values.entrySet())
+      {
+         AttributionListener.Key key = entry.getKey();
+         AttributionListener.Counters counts = entry.getValue();
+         QRecord row = new QRecord().withValue("windowStart", start).withValue("windowEnd", end)
+            .withValue("podName", truncate(runtime.getConfig().getPodName(), 200)).withValue("unitKind", key.kind())
+            .withValue("unitName", truncate(key.name(), 300)).withValue("processName", truncate(key.process(), 200))
+            .withValue("tableName", truncate(key.table(), 200)).withValue("fingerprint", key.fingerprint())
+            .withValue("sqlText", truncate(counts.sql, 4000)).withValue("sampleRate", counts.sampleRate < 0 ? null : BigDecimal.valueOf(counts.sampleRate));
+         if(repeat)
+         {
+            row.withValue("unitsAffected", counts.units.sum()).withValue("totalRepeats", counts.statements.sum())
+               .withValue("maxRepeats", (int) Math.min(Integer.MAX_VALUE, counts.max.get()));
+         }
+         else
+         {
+            row.withValue("operation", key.operation()).withValue("units", counts.units.sum()).withValue("statements", counts.statements.sum())
+               .withValue("totalMs", BigDecimal.valueOf(counts.nanos.sum(), 6)).withValue("maxMs", BigDecimal.valueOf(counts.max.get(), 6))
+               .withValue("rowsCount", counts.rows.sum()).withValue("errors", counts.errors.sum());
+         }
+         records.add(row);
+      }
+      return (records);
+   }
+
+
+   /*******************************************************************************
+    ** Bound text at the persistence boundary after normalization.
+    *******************************************************************************/
+   private static String truncate(String value, int length)
+   {
+      return (value == null || value.length() <= length ? value : value.substring(0, length));
+   }
+
+
+   /*******************************************************************************
+    ** Reject record-level failures so the whole window transaction rolls back.
+    *******************************************************************************/
+   private static void insert(String table, List<QRecord> rows) throws Exception
+   {
+      if(!rows.isEmpty())
+      {
+         List<QRecord> inserted = InsertAction.executeForRecords(new InsertInput(table).withRecords(rows).withTransaction(QContext.getQBackendTransaction()));
+         if(inserted.size() != rows.size() || inserted.stream().anyMatch(row -> row.getErrors() != null && !row.getErrors().isEmpty()))
+         {
+            throw (new IllegalStateException("Attribution insert failed"));
+         }
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Listener exposed for deterministic delivery and registration tests.
+    *******************************************************************************/
+   public AttributionListener getListener()
+   {
+      return (listener);
+   }
+
+
+   /*******************************************************************************
+    ** Unregister first, stop scheduling, and flush one final detached window.
+    *******************************************************************************/
+   @Override
+   public synchronized void close()
+   {
+      if(closed)
+      {
+         return;
+      }
+      closed = true;
+      StatementExecutionListeners.unregister(listener);
+      if(executor != null)
+      {
+         executor.shutdownNow();
+      }
+      if(shutdownHook != null && Thread.currentThread() != shutdownHook)
+      {
+         try
+         {
+            Runtime.getRuntime().removeShutdownHook(shutdownHook);
+         }
+         catch(IllegalStateException expected)
+         {
+            // JVM shutdown is already running the hook.
+         }
+      }
+      flushNow();
+   }
+}

src/main/java/com/kingsrook/qbits/dbperformance/attribution/AttributionListener.java

@@ -0,0 +1,425 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dbperformance.attribution;
+
+
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAdder;
+import com.kingsrook.qbits.dbperformance.DbPerfQBitConfig;
+import com.kingsrook.qbits.dbperformance.DbPerfRuntime;
+import com.kingsrook.qbits.dbperformance.normalize.SqlNormalizer;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.instrumentation.StatementExecution;
+import com.kingsrook.qqq.backend.core.instrumentation.StatementExecutionListener;
+import com.kingsrook.qqq.backend.core.instrumentation.UnitOfWork;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+
+
+/*******************************************************************************
+ ** Bounded, in-memory attribution. No table actions run on statement threads.
+ ** Completed units belong to the window in which they end, including long units.
+ *******************************************************************************/
+public class AttributionListener implements StatementExecutionListener
+{
+   private static final String OTHER = "(other)";
+   private static final Key OVERFLOW = new Key("OTHER", OTHER, OTHER, OTHER, "OTHER", SqlNormalizer.fingerprint(OTHER), -1);
+   private static final ThreadLocal<Boolean> SUPPRESSED = ThreadLocal.withInitial(() -> false);
+   private final DbPerfRuntime runtime;
+   private final ConcurrentHashMap<String, Shape> shapes = new ConcurrentHashMap<>();
+   private final ThreadLocal<Buffer> current = new ThreadLocal<>();
+   private volatile Settings settings = new Settings(false, 1.0, 60, 10000, 10);
+   private Window window = new Window(Instant.now(), new HashMap<>(), new HashMap<>());
+
+
+   /*******************************************************************************
+    ** Bind to one instance and monitored backend.
+    *******************************************************************************/
+   public AttributionListener(DbPerfRuntime runtime)
+   {
+      this.runtime = runtime;
+   }
+
+
+   /*******************************************************************************
+    ** Ignore self-observation, then sample and count without database access.
+    *******************************************************************************/
+   @Override
+   public void onStatement(StatementExecution execution)
+   {
+      Settings copy = settings;
+      if(!copy.enabled() || !Boolean.TRUE.equals(runtime.getConfig().getAttributionEnabled()) || SUPPRESSED.get()
+         || !runtime.getConfig().getBackendName().equals(execution.backendName())
+         || (execution.tableName() != null && execution.tableName().startsWith(DbPerfQBitConfig.TABLE_NAME_PREFIX))
+         || (QContext.getQInstance() != null && QContext.getQInstance() != runtime.getInstance()))
+      {
+         return;
+      }
+      UnitOfWork unit = execution.unitOfWork();
+      Buffer buffer = current.get();
+      if(buffer != null && (unit == null || buffer.unit.getId() != unit.getId()))
+      {
+         merge(buffer);
+         current.remove();
+         buffer = null;
+      }
+      if(unit != null && buffer == null)
+      {
+         buffer = new Buffer(unit, copy, sampled(unit.getId(), copy.sampleRate()));
+         current.set(buffer);
+      }
+      if(buffer == null ? !sampled(ThreadLocalRandom.current().nextLong(), copy.sampleRate()) : !buffer.sampled)
+      {
+         return;
+      }
+      Shape shape = shape(execution.sql());
+      Key key = new Key(unit == null ? "OTHER" : unit.getKind().name(), unit == null ? name(execution.processName()) : "",
+         execution.processName(), execution.tableName(), execution.operation() == null ? "OTHER" : execution.operation().name(), shape.fingerprint(), buffer == null ? copy.sampleRate() : buffer.settings.sampleRate());
+      if(buffer == null)
+      {
+         synchronized(this)
+         {
+            Key target = bounded(window.attribution(), key, copy.maxKeys());
+            Counters counts = window.attribution().computeIfAbsent(target, ignored -> new Counters(target == OVERFLOW ? OTHER : shape.sql(), key.sampleRate()));
+            counts.sampleRate(key.sampleRate());
+            counts.add(execution);
+            counts.units.increment();
+         }
+      }
+      else
+      {
+         Key target = bounded(buffer.counts, key, Math.max(2, buffer.settings.maxKeys()));
+         buffer.counts.computeIfAbsent(target, ignored -> new Counters(target == OVERFLOW ? OTHER : shape.sql(), key.sampleRate())).add(execution);
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Read the final route name, then release all per-thread counters.
+    *******************************************************************************/
+   @Override
+   public void onUnitOfWorkEnd(UnitOfWork unit)
+   {
+      Buffer buffer = current.get();
+      if(buffer != null && buffer.unit.getId() == unit.getId())
+      {
+         current.remove();
+         if(settings.enabled() && Boolean.TRUE.equals(runtime.getConfig().getAttributionEnabled()))
+         {
+            merge(buffer);
+         }
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Merge each unit once per destination, even when several shapes overflow.
+    ** Repeat counts are combined across operations before threshold evaluation.
+    *******************************************************************************/
+   private synchronized void merge(Buffer buffer)
+   {
+      Set<Key> touched = new HashSet<>();
+      Map<Key, Counters> repeats = new HashMap<>();
+      for(Map.Entry<Key, Counters> entry : buffer.counts.entrySet())
+      {
+         Key local = entry.getKey();
+         Key key = local == OVERFLOW ? OVERFLOW : new Key(local.kind(), name(buffer.unit.getName()), local.process(), local.table(), local.operation(), local.fingerprint(), buffer.settings.sampleRate());
+         Key target = bounded(window.attribution(), key, settings.maxKeys());
+         Counters counts = window.attribution().computeIfAbsent(target, ignored -> new Counters(target == OVERFLOW ? OTHER : entry.getValue().sql, entry.getValue().sampleRate));
+         counts.merge(entry.getValue());
+         if(touched.add(target))
+         {
+            counts.units.increment();
+         }
+         // Collapsed shapes cannot prove repetition of any one fingerprint.
+         if(local != OVERFLOW)
+         {
+            Key repeat = new Key(key.kind(), key.name(), key.process(), key.table(), null, key.fingerprint(), key.sampleRate());
+            repeats.computeIfAbsent(repeat, ignored -> new Counters(entry.getValue().sql, entry.getValue().sampleRate)).merge(entry.getValue());
+         }
+      }
+      Map<Key, Long> totals = new HashMap<>();
+      Map<Key, Long> maxima = new HashMap<>();
+      for(Map.Entry<Key, Counters> entry : repeats.entrySet())
+      {
+         long count = entry.getValue().statements.sum();
+         if(count >= buffer.settings.threshold())
+         {
+            Key target = bounded(window.repeats(), entry.getKey(), settings.maxKeys());
+            window.repeats().computeIfAbsent(target, ignored -> new Counters(target == OVERFLOW ? OTHER : entry.getValue().sql, entry.getValue().sampleRate));
+            window.repeats().get(target).sampleRate(entry.getValue().sampleRate);
+            totals.merge(target, count, Long::sum);
+            maxima.merge(target, count, Math::max);
+         }
+      }
+      for(Map.Entry<Key, Long> entry : totals.entrySet())
+      {
+         Counters counts = window.repeats().get(entry.getKey());
+         counts.units.increment();
+         counts.statements.add(entry.getValue());
+         counts.max.accumulateAndGet(maxima.get(entry.getKey()), Math::max);
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Reserve one slot for a universal overflow key, preserving the strict cap.
+    *******************************************************************************/
+   private static Key bounded(Map<Key, Counters> counts, Key key, int limit)
+   {
+      return (counts.containsKey(key) || (key != OVERFLOW && counts.size() < limit - (counts.containsKey(OVERFLOW) ? 0 : 1)) ? key : OVERFLOW);
+   }
+
+
+   /*******************************************************************************
+    ** Normalize once per raw SQL shape; cold-cache admission alone is serialized.
+    *******************************************************************************/
+   private Shape shape(String sql)
+   {
+      String raw = sql == null ? "" : sql;
+      Shape cached = shapes.get(raw);
+      if(cached != null)
+      {
+         return (cached);
+      }
+      String normalized = SqlNormalizer.normalize(raw);
+      Shape created = new Shape(normalized, SqlNormalizer.fingerprint(normalized));
+      synchronized(shapes)
+      {
+         if(shapes.size() >= 5000)
+         {
+            shapes.clear();
+         }
+         shapes.put(raw, created);
+      }
+      return (created);
+   }
+
+
+   /*******************************************************************************
+    ** SplitMix64 finalizer distributes sequential unit IDs without allocations.
+    *******************************************************************************/
+   private static boolean sampled(long id, double rate)
+   {
+      long hash = (id ^ (id >>> 30)) * 0xbf58476d1ce4e5b9L;
+      hash = (hash ^ (hash >>> 27)) * 0x94d049bb133111ebL;
+      hash ^= hash >>> 31;
+      return ((hash >>> 11) * 0x1.0p-53 < rate);
+   }
+
+
+   /*******************************************************************************
+    ** Use a stable label for units without a name.
+    *******************************************************************************/
+   private static String name(String value)
+   {
+      return (value == null || value.isBlank() ? "(none)" : value);
+   }
+
+
+   /*******************************************************************************
+    ** Detach an immutable-in-practice window; no writers retain its counters.
+    *******************************************************************************/
+   synchronized Window drain(Instant end)
+   {
+      Window previous = window;
+      window = new Window(end, new HashMap<>(), new HashMap<>());
+      return (previous);
+   }
+
+
+   /*******************************************************************************
+    ** Publish a validated immutable settings copy only from the flusher.
+    *******************************************************************************/
+   void refresh(QRecord row)
+   {
+      double rate = row.getValueBigDecimal("attributionSampleRate") == null ? 1.0 : row.getValueBigDecimal("attributionSampleRate").doubleValue();
+      settings = new Settings(!Boolean.FALSE.equals(row.getValueBoolean("attributionEnabled")), Math.max(0, Math.min(1, rate)),
+         positive(row, "attributionFlushSeconds", 60), positive(row, "attributionMaxKeys", 10000), positive(row, "nPlusOneThreshold", 10));
+   }
+
+
+   /*******************************************************************************
+    ** Invalid administrative values use defaults.
+    *******************************************************************************/
+   private static int positive(QRecord row, String field, int fallback)
+   {
+      Integer value = row.getValueInteger(field);
+      return (value == null || value <= 0 ? fallback : value);
+   }
+
+
+   /*******************************************************************************
+    ** Current settings for scheduling and window annotations.
+    *******************************************************************************/
+   Settings settings()
+   {
+      return (settings);
+   }
+
+
+   /*******************************************************************************
+    ** Set thread suppression, returning the previous state for nested callers.
+    *******************************************************************************/
+   static boolean suppress(boolean value)
+   {
+      boolean previous = SUPPRESSED.get();
+      if(value)
+      {
+         SUPPRESSED.set(true);
+      }
+      else
+      {
+         SUPPRESSED.remove();
+      }
+      return (previous);
+   }
+
+
+   /*******************************************************************************
+    ** Immutable administrative settings.
+    *******************************************************************************/
+   record Settings(boolean enabled, double sampleRate, int flushSeconds, int maxKeys, int threshold)
+   {
+   }
+
+
+   /*******************************************************************************
+    ** Attribution dimensions; sampling transitions remain separately interpretable.
+    ** Null operation denotes a repeat pattern.
+    *******************************************************************************/
+   record Key(String kind, String name, String process, String table, String operation, String fingerprint, double sampleRate)
+   {
+   }
+
+
+   /*******************************************************************************
+    ** Cached literal-free text and its digest.
+    *******************************************************************************/
+   private record Shape(String sql, String fingerprint)
+   {
+   }
+
+
+   /*******************************************************************************
+    ** Detached window transferred to the flusher.
+    *******************************************************************************/
+   record Window(Instant start, Map<Key, Counters> attribution, Map<Key, Counters> repeats)
+   {
+   }
+
+
+   /*******************************************************************************
+    ** Thread-confined unit, including its once-only sampling decision.
+    *******************************************************************************/
+   private static class Buffer
+   {
+      private final UnitOfWork unit;
+      private final Settings settings;
+      private final boolean sampled;
+      private final Map<Key, Counters> counts = new HashMap<>();
+
+
+      /*******************************************************************************
+       ** Capture unit settings so refresh cannot partially sample a unit.
+       *******************************************************************************/
+      private Buffer(UnitOfWork unit, Settings settings, boolean sampled)
+      {
+         this.unit = unit;
+         this.settings = settings;
+         this.sampled = sampled;
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Additive counters plus an atomic maximum, shared by both window tables.
+    *******************************************************************************/
+   static class Counters
+   {
+      final String sql;
+      double sampleRate;
+      final LongAdder units = new LongAdder();
+      final LongAdder statements = new LongAdder();
+      final LongAdder nanos = new LongAdder();
+      final AtomicLong max = new AtomicLong();
+      final LongAdder rows = new LongAdder();
+      final LongAdder errors = new LongAdder();
+
+
+      /*******************************************************************************
+       ** Retain only normalized SQL.
+       *******************************************************************************/
+      Counters(String sql, double sampleRate)
+      {
+         this.sql = sql;
+         this.sampleRate = sampleRate;
+      }
+
+
+      /*******************************************************************************
+       ** Count one completed statement, including failures and unknown row counts.
+       *******************************************************************************/
+      void add(StatementExecution execution)
+      {
+         statements.increment();
+         nanos.add(execution.elapsedNanos());
+         max.accumulateAndGet(execution.elapsedNanos(), Math::max);
+         rows.add(execution.rowCount() == null ? 0 : execution.rowCount());
+         if(!execution.succeeded())
+         {
+            errors.increment();
+         }
+      }
+
+
+      /*******************************************************************************
+       ** An overflow bucket spanning sampling rates has no single valid rate.
+       *******************************************************************************/
+      void sampleRate(double rate)
+      {
+         if(sampleRate != rate)
+         {
+            sampleRate = -1;
+         }
+      }
+
+
+      /*******************************************************************************
+       ** Fold statement totals; distinct units are managed by the caller.
+       *******************************************************************************/
+      void merge(Counters other)
+      {
+         sampleRate(other.sampleRate);
+         statements.add(other.statements.sum());
+         nanos.add(other.nanos.sum());
+         max.accumulateAndGet(other.max.get(), Math::max);
+         rows.add(other.rows.sum());
+         errors.add(other.errors.sum());
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/dbperformance/attribution/AttributionRuntimeService.java

@@ -0,0 +1,69 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dbperformance.attribution;
+
+
+import com.kingsrook.qbits.dbperformance.DbPerfRuntime;
+import com.kingsrook.qqq.backend.core.instances.QRuntimeServiceInterface;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+
+
+/*******************************************************************************
+ ** QApplicationLauncher lifecycle, after instance initialization and migrations.
+ *******************************************************************************/
+public class AttributionRuntimeService implements QRuntimeServiceInterface
+{
+   private DbPerfRuntime runtime;
+
+
+   /*******************************************************************************
+    ** Name used by the host launcher.
+    *******************************************************************************/
+   @Override
+   public String getName()
+   {
+      return ("dbPerformanceAttribution");
+   }
+
+
+   /*******************************************************************************
+    ** Start on each application pod, independently of the job scheduler.
+    *******************************************************************************/
+   @Override
+   public void start(QInstance instance)
+   {
+      runtime = DbPerfRuntime.of(instance);
+      runtime.startAttribution();
+   }
+
+
+   /*******************************************************************************
+    ** Stop listener delivery and persist the final completed units.
+    *******************************************************************************/
+   @Override
+   public void stop()
+   {
+      if(runtime != null)
+      {
+         runtime.stopAttribution();
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/dbperformance/metadata/DbPerfMetaData.java

@@ -549,9 +549,10 @@ public final class DbPerfMetaData
       repeatPattern.getField("unitsAffected").setBackendName("units_affected");
       repeatPattern.getField("maxRepeats").setBackendName("max_repeats");
       repeatPattern.getField("totalRepeats").setBackendName("total_repeats");
+      repeatPattern.getField("sampleRate").setBackendName("sample_rate");
       repeatPattern.withoutCapabilities(Capability.TABLE_INSERT, Capability.TABLE_UPDATE, Capability.TABLE_DELETE);
       repeatPattern.withSection(new QFieldSection("identity", new QIcon("table_rows"), Tier.T1, List.of("id", "windowStart", "windowEnd", "podName", "unitKind", "unitName", "processName")));
-      repeatPattern.withSection(new QFieldSection("details", new QIcon("table_rows"), Tier.T2, List.of("tableName", "fingerprint", "sqlText", "unitsAffected", "maxRepeats", "totalRepeats")));
+      repeatPattern.withSection(new QFieldSection("details", new QIcon("table_rows"), Tier.T2, List.of("tableName", "fingerprint", "sqlText", "unitsAffected", "maxRepeats", "totalRepeats", "sampleRate")));
       instance.addTable(repeatPattern);
 
       QTableMetaData finding = base(config, DbPerfFinding.TABLE_NAME, "Findings", DbPerfFinding.class);

src/main/java/com/kingsrook/qbits/dbperformance/model/DbPerfRepeatPattern.java

@@ -21,6 +21,7 @@
 package com.kingsrook.qbits.dbperformance.model;
 
 
+import java.math.BigDecimal;
 import java.time.Instant;
 import com.kingsrook.qqq.backend.core.model.data.QField;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
@@ -73,6 +74,9 @@ public class DbPerfRepeatPattern extends QRecordEntity
    @QField
    private Long totalRepeats;
 
+   @QField
+   private BigDecimal sampleRate;
+
 
    /*******************************************************************************
     ** Default constructor.
@@ -453,4 +457,31 @@ public class DbPerfRepeatPattern extends QRecordEntity
       this.totalRepeats = totalRepeats;
       return (this);
    }
+
+   /*******************************************************************************
+    ** Getter for sampleRate.
+    *******************************************************************************/
+   public BigDecimal getSampleRate()
+   {
+      return (sampleRate);
+   }
+
+
+   /*******************************************************************************
+    ** Setter for sampleRate.
+    *******************************************************************************/
+   public void setSampleRate(BigDecimal sampleRate)
+   {
+      this.sampleRate = sampleRate;
+   }
+
+
+   /*******************************************************************************
+    ** Fluent setter for sampleRate.
+    *******************************************************************************/
+   public DbPerfRepeatPattern withSampleRate(BigDecimal sampleRate)
+   {
+      this.sampleRate = sampleRate;
+      return (this);
+   }
 }

src/main/resources/db/liquibase/qbit-db-performance/db_performance.changelog.sql

@@ -605,3 +605,7 @@ INSERT INTO db_perf_rule (code,label,description,enabled,severity,threshold_1,th
 SELECT 'TEMP_SPILL', 'Temporary disk spill', 'Statements that write temporary files because sorts or hashes do not fit in work_mem.', TRUE, 'WARNING', 100, 'Temporary data written per day (MB)', NULL, NULL WHERE NOT EXISTS (SELECT 1 FROM db_perf_rule WHERE code = 'TEMP_SPILL');
 INSERT INTO db_perf_rule (code,label,description,enabled,severity,threshold_1,threshold_1_label,threshold_2,threshold_2_label)
 SELECT 'DEADLOCKS', 'Deadlocks', 'Transactions that deadlocked and were cancelled by PostgreSQL. Usually two code paths that lock rows in a different order.', TRUE, 'WARNING', 1, 'Deadlocks per day', NULL, NULL WHERE NOT EXISTS (SELECT 1 FROM db_perf_rule WHERE code = 'DEADLOCKS');
+
+-- changeset qbit-db-performance:34
+-- comment: Repeated Statements: the sampling rate of each window, so repeat counts can be read at scale.
+ALTER TABLE db_perf_repeat_pattern ADD COLUMN IF NOT EXISTS sample_rate DECIMAL(5,4);

src/main/resources/help/qbit-db-performance.yaml

@@ -388,6 +388,8 @@ entries:
   summary: "Largest repetition count for one fingerprint in a single sampled unit of work."
 - key: table:db_perf_repeat_pattern;field:totalRepeats
   summary: "Total repeats reported for this repeat pattern record."
+- key: table:db_perf_repeat_pattern;field:sampleRate
+  summary: "Share of units of work that were measured in this window (1 means all)."
 - key: table:db_perf_rule;slot:help
   summary: "Editable thresholds and severity for each suggestion rule."
   content: |

src/test/java/com/kingsrook/qbits/dbperformance/RollupRetentionTest.java

@@ -112,6 +112,9 @@ class RollupRetentionTest
       }
       rows.add(new QRecord().withValue("windowStart", now.minusSeconds(2 * 86400)).withValue("unitName", "boundary"));
       DbPerfStore.insert("db_perf_attribution", rows);
+      DbPerfStore.insert("db_perf_repeat_pattern", List.of(
+         new QRecord().withValue("windowStart", now.minusSeconds(3 * 86400)),
+         new QRecord().withValue("windowStart", now.minusSeconds(2 * 86400))));
       DbPerfStore.insert("db_perf_attribution_daily", List.of(new QRecord().withValue("day", LocalDate.of(2026, 9, 20)).withValue("rollupKey", "old")));
       int rule = DbPerfStore.query("db_perf_rule", new QQueryFilter()).getFirst().getValueInteger("id");
       List<QRecord> findings = new ArrayList<>();
@@ -121,8 +124,9 @@ class RollupRetentionTest
       }
       DbPerfStore.insert("db_perf_finding", findings);
       RetentionService retention = new RetentionService(runtime, Clock.fixed(now, ZoneOffset.UTC));
-      assertThat(retention.purge()).isEqualTo(1008);
+      assertThat(retention.purge()).isEqualTo(1009);
       assertThat(retention.purge()).isZero();
+      assertThat(DbPerfStore.query("db_perf_repeat_pattern", new QQueryFilter())).hasSize(1);
       assertThat(DbPerfStore.query("db_perf_attribution", new QQueryFilter())).singleElement().satisfies(row -> assertThat(row.getValueString("unitName")).isEqualTo("boundary"));
       assertThat(DbPerfStore.query("db_perf_finding", new QQueryFilter())).extracting(row -> row.getValueString("status")).containsExactlyInAnyOrder("OPEN", "ACKNOWLEDGED");
    }

src/test/java/com/kingsrook/qbits/dbperformance/attribution/AttributionTest.java

@@ -0,0 +1,645 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dbperformance.attribution;
+
+
+import java.math.BigDecimal;
+import java.sql.DriverManager;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import com.kingsrook.qbits.dbperformance.DbPerfRuntime;
+import com.kingsrook.qbits.dbperformance.DbPerfTestSupport;
+import com.kingsrook.qbits.dbperformance.collect.DbPerfStore;
+import com.kingsrook.qbits.dbperformance.collect.RollupService;
+import com.kingsrook.qbits.dbperformance.metadata.DbPerfWriteGuard;
+import com.kingsrook.qbits.dbperformance.normalize.SqlNormalizer;
+import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.instrumentation.StatementExecution;
+import com.kingsrook.qqq.backend.core.instrumentation.StatementExecutionListener;
+import com.kingsrook.qqq.backend.core.instrumentation.StatementExecutionListeners;
+import com.kingsrook.qqq.backend.core.instrumentation.StatementOperation;
+import com.kingsrook.qqq.backend.core.instrumentation.UnitOfWork;
+import com.kingsrook.qqq.backend.core.instrumentation.UnitOfWorkKind;
+import com.kingsrook.qqq.backend.core.instrumentation.UnitsOfWork;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData;
+import com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSTableBackendDetails;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/*******************************************************************************
+ ** Real QQQ RDBMS hooks, persistence, bounded counters, and lifecycle contracts.
+ *******************************************************************************/
+class AttributionTest
+{
+   private DbPerfRuntime runtime;
+   private AttributionFlusher flusher;
+   private AttributionListener listener;
+   private QRecord settings;
+
+
+   /*******************************************************************************
+    ** A real business table, separate from the self-suppressed db_perf tables.
+    *******************************************************************************/
+   @BeforeEach
+   void setUp() throws Exception
+   {
+      QInstance instance = DbPerfTestSupport.instance();
+      runtime = DbPerfRuntime.of(instance);
+      RDBMSBackendMetaData backend = (RDBMSBackendMetaData) instance.getBackend(runtime.getConfig().getBackendName());
+      instance.addTable(new QTableMetaData().withName("business").withBackendName(backend.getName()).withPrimaryKeyField("id")
+         .withBackendDetails(new RDBMSTableBackendDetails().withTableName("business"))
+         .withField(new QFieldMetaData("id", QFieldType.INTEGER)).withField(new QFieldMetaData("name", QFieldType.STRING)));
+      try(var connection = DriverManager.getConnection(backend.getJdbcUrl(), backend.getUsername(), backend.getPassword()); var statement = connection.createStatement())
+      {
+         statement.execute("CREATE TABLE business (id INT PRIMARY KEY, name VARCHAR(100))");
+         statement.execute("INSERT INTO business VALUES (1, 'example')");
+      }
+      flusher = new AttributionFlusher(runtime);
+      flusher.refreshSettings();
+      listener = flusher.getListener();
+      settings = DbPerfStore.query("db_perf_settings", new QQueryFilter()).getFirst();
+      StatementExecutionListeners.register(listener);
+   }
+
+
+   /*******************************************************************************
+    ** No listener, scope, suppression or runtime may escape a test.
+    *******************************************************************************/
+   @AfterEach
+   void tearDown()
+   {
+      StatementExecutionListeners.unregister(listener);
+      flusher.close();
+      runtime.stopAttribution();
+      UnitsOfWork.reset();
+      AttributionListener.suppress(false);
+      QContext.clear();
+   }
+
+
+   /*******************************************************************************
+    ** The core hook attributes real queries and reads final HTTP route names.
+    *******************************************************************************/
+   @Test
+   void realQueriesAndFinalNames() throws Exception
+   {
+      try(var scope = UnitsOfWork.begin(UnitOfWorkKind.SCHEDULED_JOB, "job"))
+      {
+         query(2);
+      }
+      try(var scope = UnitsOfWork.begin(UnitOfWorkKind.HTTP_REQUEST, "unresolved"))
+      {
+         query(1);
+         scope.getUnitOfWork().setName("GET /things/{id}");
+      }
+      query(1);
+      flusher.flushNow();
+      List<QRecord> rows = rows("db_perf_attribution");
+      assertThat(rows).hasSize(3);
+      assertThat(rows).extracting(row -> row.getValueString("unitName")).containsExactlyInAnyOrder("job", "GET /things/{id}", "(none)");
+      QRecord job = named(rows, "job");
+      assertThat(job.getValueString("unitKind")).isEqualTo("SCHEDULED_JOB");
+      assertThat(job.getValueLong("units")).isEqualTo(1);
+      assertThat(job.getValueLong("statements")).isEqualTo(2);
+      assertThat(job.getValueLong("rowsCount")).isEqualTo(2);
+      assertThat(job.getValueString("operation")).isEqualTo("QUERY");
+      assertThat(job.getValueString("tableName")).isEqualTo("business");
+      assertThat(job.getValueString("fingerprint")).isEqualTo(SqlNormalizer.fingerprint(job.getValueString("sqlText")));
+      assertThat(named(rows, "(none)").getValueString("unitKind")).isEqualTo("OTHER");
+   }
+
+
+   /*******************************************************************************
+    ** The threshold applies to each unit, and each window uses one batch/table.
+    *******************************************************************************/
+   @Test
+   void repeatPatternsBatchesAndSelfSuppression() throws Exception
+   {
+      AtomicInteger attributionInserts = new AtomicInteger();
+      AtomicInteger repeatInserts = new AtomicInteger();
+      StatementExecutionListener observer = execution ->
+      {
+         if(execution.operation() == StatementOperation.INSERT)
+         {
+            if("db_perf_attribution".equals(execution.tableName()))
+            {
+               attributionInserts.incrementAndGet();
+            }
+            if("db_perf_repeat_pattern".equals(execution.tableName()))
+            {
+               repeatInserts.incrementAndGet();
+            }
+         }
+      };
+      StatementExecutionListeners.register(observer);
+      try
+      {
+         for(int count : List.of(12, 5))
+         {
+            try(var scope = UnitsOfWork.begin(UnitOfWorkKind.SCHEDULED_JOB, "job"))
+            {
+               query(count);
+            }
+         }
+         rows("db_perf_settings");
+         flusher.flushNow();
+         QRecord repeated = rows("db_perf_repeat_pattern").getFirst();
+         assertThat(repeated.getValueLong("unitsAffected")).isEqualTo(1);
+         assertThat(repeated.getValueInteger("maxRepeats")).isEqualTo(12);
+         assertThat(repeated.getValueLong("totalRepeats")).isEqualTo(12);
+         assertThat(repeated.getValueBigDecimal("sampleRate")).isEqualByComparingTo("1");
+         assertThat(rows("db_perf_repeat_pattern")).hasSize(1);
+         assertThat(rows("db_perf_attribution")).singleElement().satisfies(row ->
+         {
+            assertThat(row.getValueLong("units")).isEqualTo(2);
+            assertThat(row.getValueLong("statements")).isEqualTo(17);
+         });
+         flusher.flushNow();
+         assertThat(attributionInserts).hasValue(1);
+         assertThat(repeatInserts).hasValue(1);
+         assertThat(listener.drain(Instant.now()).attribution()).isEmpty();
+      }
+      finally
+      {
+         StatementExecutionListeners.unregister(observer);
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Both runtime switches and sample zero take effect through cached settings.
+    *******************************************************************************/
+   @Test
+   void disabledSamplingAndFilters() throws Exception
+   {
+      settings.withValue("attributionSampleRate", BigDecimal.ZERO);
+      refresh();
+      query(1);
+      try(var scope = UnitsOfWork.begin(UnitOfWorkKind.PROCESS, "zero"))
+      {
+         query(1);
+      }
+      assertThat(listener.drain(Instant.now()).attribution()).isEmpty();
+      settings.withValue("attributionSampleRate", BigDecimal.ONE).withValue("attributionEnabled", false);
+      refresh();
+      query(1);
+      assertThat(listener.drain(Instant.now()).attribution()).isEmpty();
+      settings.withValue("attributionEnabled", true);
+      refresh();
+      runtime.getConfig().setAttributionEnabled(false);
+      query(1);
+      runtime.getConfig().setAttributionEnabled(true);
+      listener.onStatement(new StatementExecution("wrong", "business", StatementOperation.QUERY, "select ?", 1, 1, true, null, null));
+      AttributionListener.suppress(true);
+      query(1);
+      AttributionListener.suppress(false);
+      rows("db_perf_settings");
+      assertThat(listener.drain(Instant.now()).attribution()).isEmpty();
+      query(1);
+      assertThat(listener.drain(Instant.now()).attribution()).hasSize(1);
+   }
+
+
+   /*******************************************************************************
+    ** The cap includes the overflow bucket; a unit counts only once in it.
+    *******************************************************************************/
+   @Test
+   void overflowAndRepeatBounds() throws Exception
+   {
+      settings.withValue("attributionMaxKeys", 2).withValue("nPlusOneThreshold", 2);
+      refresh();
+      try(var scope = UnitsOfWork.begin(UnitOfWorkKind.PROCESS, "many-shapes"))
+      {
+         for(int i = 0; i < 8; i++)
+         {
+            synthetic("select column_" + i + " from business", 1, true, scope.getUnitOfWork());
+         }
+      }
+      for(int i = 0; i < 8; i++)
+      {
+         try(var scope = UnitsOfWork.begin(UnitOfWorkKind.PROCESS, "job_" + i))
+         {
+            synthetic("select column_" + i + " from business", 1, true, scope.getUnitOfWork());
+            synthetic("select column_" + i + " from business", 1, true, scope.getUnitOfWork());
+         }
+      }
+      flusher.flushNow();
+      assertThat(rows("db_perf_attribution")).hasSize(2);
+      assertThat(named(rows("db_perf_attribution"), "(other)").getValueLong("units")).isEqualTo(9);
+      assertThat(rows("db_perf_attribution").stream().mapToLong(row -> row.getValueLong("statements")).sum()).isEqualTo(24);
+      assertThat(rows("db_perf_repeat_pattern")).hasSize(2);
+      assertThat(named(rows("db_perf_repeat_pattern"), "(other)").getValueString("fingerprint")).isEqualTo(SqlNormalizer.fingerprint("(other)"));
+   }
+
+
+   /*******************************************************************************
+    ** Counters, normalization, null rows, process fallback, and text bounds.
+    *******************************************************************************/
+   @Test
+   void syntheticCountersAndSqlSafety()
+   {
+      String sql = "select 'secret', 123 from business where id in (?, ?, ?)";
+      synthetic(sql, 2000000, true, null);
+      synthetic(sql, 5000000, false, null);
+      listener.onStatement(new StatementExecution(runtime.getConfig().getBackendName(), null, null, null, 10, null, true, null, "process"));
+      flusher.flushNow();
+      List<QRecord> rows = rows("db_perf_attribution");
+      QRecord row = named(rows, "(none)");
+      assertThat(row.getValueBigDecimal("totalMs")).isEqualByComparingTo("7");
+      assertThat(row.getValueBigDecimal("maxMs")).isEqualByComparingTo("5");
+      assertThat(row.getValueLong("errors")).isEqualTo(1);
+      assertThat(row.getValueLong("rowsCount")).isEqualTo(2);
+      assertThat(row.getValueString("sqlText")).doesNotContain("secret", "123");
+      assertThat(named(rows, "process").getValueString("unitKind")).isEqualTo("OTHER");
+      synthetic("select " + "identifier".repeat(600) + " from business", 1, true, null);
+      flusher.flushNow();
+      assertThat(rows("db_perf_attribution")).anySatisfy(value -> assertThat(value.getValueString("sqlText")).hasSize(4000));
+   }
+
+
+   /*******************************************************************************
+    ** An abandoned unit is merged when another unit arrives on that worker.
+    *******************************************************************************/
+   @Test
+   void defensiveUnitTransitionAndWindowBoundary() throws Exception
+   {
+      var abandoned = UnitsOfWork.begin(UnitOfWorkKind.PROCESS, "abandoned");
+      query(1);
+      UnitsOfWork.reset();
+      try(var scope = UnitsOfWork.begin(UnitOfWorkKind.HTTP_REQUEST, "initial"))
+      {
+         query(1);
+         flusher.flushNow();
+         assertThat(rows("db_perf_attribution")).singleElement().satisfies(row -> assertThat(row.getValueString("unitName")).isEqualTo("abandoned"));
+         scope.getUnitOfWork().setName("final");
+         query(1);
+      }
+      abandoned.close();
+      flusher.flushNow();
+      assertThat(named(rows("db_perf_attribution"), "final").getValueLong("statements")).isEqualTo(2);
+   }
+
+
+   /*******************************************************************************
+    ** Each sampled unit contributes all statements, including across a refresh.
+    *******************************************************************************/
+   @Test
+   void samplingIsPerUnit() throws Exception
+   {
+      settings.withValue("attributionSampleRate", new BigDecimal("0.5"));
+      refresh();
+      for(int i = 0; i < 100; i++)
+      {
+         try(var scope = UnitsOfWork.begin(UnitOfWorkKind.PROCESS, "sample_" + i))
+         {
+            for(int j = 0; j < 3; j++)
+            {
+               synthetic("select ?", 1, true, scope.getUnitOfWork());
+            }
+         }
+      }
+      var window = listener.drain(Instant.now());
+      assertThat(window.attribution().size()).isBetween(20, 80);
+      assertThat(window.attribution().values()).allSatisfy(counts -> assertThat(counts.statements.sum()).isEqualTo(3));
+   }
+
+
+   /*******************************************************************************
+    ** An in-flight unit keeps its sampling decision and original rate annotation.
+    *******************************************************************************/
+   @Test
+   void samplingRefreshDuringUnit() throws Exception
+   {
+      try(var scope = UnitsOfWork.begin(UnitOfWorkKind.PROCESS, "in-flight"))
+      {
+         query(1);
+         settings.withValue("attributionSampleRate", BigDecimal.ZERO);
+         refresh();
+         query(1);
+      }
+      query(1);
+      flusher.flushNow();
+      assertThat(rows("db_perf_attribution")).singleElement().satisfies(row ->
+      {
+         assertThat(row.getValueLong("statements")).isEqualTo(2);
+         assertThat(row.getValueBigDecimal("sampleRate")).isEqualByComparingTo("1");
+      });
+   }
+
+
+   /*******************************************************************************
+    ** Atomic drains cannot lose or double-count concurrent statement delivery.
+    *******************************************************************************/
+   @Test
+   void concurrentDrains() throws Exception
+   {
+      var pool = Executors.newFixedThreadPool(4);
+      try
+      {
+         List<java.util.concurrent.Future<?>> tasks = new ArrayList<>();
+         for(int i = 0; i < 4; i++)
+         {
+            tasks.add(pool.submit(() ->
+            {
+               for(int j = 0; j < 5000; j++)
+               {
+                  synthetic("select ?", 1, true, null);
+               }
+            }));
+         }
+         long total = 0;
+         while(tasks.stream().anyMatch(task -> !task.isDone()))
+         {
+            total += listener.drain(Instant.now()).attribution().values().stream().mapToLong(counts -> counts.statements.sum()).sum();
+            Thread.yield();
+         }
+         for(var task : tasks)
+         {
+            task.get();
+         }
+         total += listener.drain(Instant.now()).attribution().values().stream().mapToLong(counts -> counts.statements.sum()).sum();
+         assertThat(total).isEqualTo(20000);
+      }
+      finally
+      {
+         pool.shutdownNow();
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Cached hot-path overhead stays below a generous two-second budget.
+    *******************************************************************************/
+   @Test
+   void overhead()
+   {
+      StatementExecution execution = new StatementExecution(runtime.getConfig().getBackendName(), "business", StatementOperation.QUERY, "select ?", 1, 1, true, null, null);
+      listener.onStatement(execution);
+      long start = System.nanoTime();
+      for(int i = 0; i < 100000; i++)
+      {
+         listener.onStatement(execution);
+      }
+      assertThat(System.nanoTime() - start).isLessThan(TimeUnit.SECONDS.toNanos(2));
+      assertThat(listener.drain(Instant.now()).attribution().values().iterator().next().statements.sum()).isEqualTo(100001);
+   }
+
+
+   /*******************************************************************************
+    ** Launcher lifecycle registers once, flushes on stop, and respects config.
+    *******************************************************************************/
+   @Test
+   void runtimeLifecycle() throws Exception
+   {
+      flusher.close();
+      assertThat(StatementExecutionListeners.isActive()).isFalse();
+      AttributionRuntimeService service = new AttributionRuntimeService();
+      assertThat(service.getName()).isEqualTo("dbPerformanceAttribution");
+      service.stop();
+      service.start(runtime.getInstance());
+      runtime.startAttribution();
+      assertThat(StatementExecutionListeners.isActive()).isTrue();
+      query(1);
+      service.stop();
+      service.stop();
+      assertThat(StatementExecutionListeners.isActive()).isFalse();
+      assertThat(rows("db_perf_attribution")).hasSize(1);
+      runtime.getConfig().setAttributionEnabled(false);
+      service.start(runtime.getInstance());
+      assertThat(StatementExecutionListeners.isActive()).isFalse();
+      service.stop();
+   }
+
+
+   /*******************************************************************************
+    ** Real flusher rows roll up across pods, preserving max and additive totals.
+    *******************************************************************************/
+   @Test
+   void rollupMergesPods() throws Exception
+   {
+      for(int i = 1; i <= 2; i++)
+      {
+         runtime.getConfig().setPodName("pod_" + i);
+         try(var scope = UnitsOfWork.begin(UnitOfWorkKind.PROCESS, "job"))
+         {
+            synthetic("select ?", i * 1000000L, true, scope.getUnitOfWork());
+            synthetic("select ?", i * 1000000L, false, scope.getUnitOfWork());
+         }
+         flusher.flushNow();
+      }
+      new RollupService(runtime).rollupDay(LocalDate.now(ZoneOffset.UTC));
+      assertThat(rows("db_perf_attribution_daily")).singleElement().satisfies(row ->
+      {
+         assertThat(row.getValueLong("units")).isEqualTo(2);
+         assertThat(row.getValueLong("statements")).isEqualTo(4);
+         assertThat(row.getValueBigDecimal("totalMs")).isEqualByComparingTo("6");
+         assertThat(row.getValueBigDecimal("maxMs")).isEqualByComparingTo("2");
+         assertThat(row.getValueLong("rowsCount")).isEqualTo(4);
+         assertThat(row.getValueLong("errors")).isEqualTo(2);
+      });
+   }
+
+
+   /*******************************************************************************
+    ** Large windows use one QQQ insert action even when the backend pages SQL.
+    *******************************************************************************/
+   @Test
+   void largeWindowUsesOneInsertAction() throws Exception
+   {
+      CountingInsert.inputs.clear();
+      for(String table : List.of("db_perf_attribution", "db_perf_repeat_pattern"))
+      {
+         runtime.getInstance().getTable(table).getCustomizers().put(TableCustomizers.PRE_INSERT_RECORD.getRole(), new QCodeReference(CountingInsert.class));
+      }
+      settings.withValue("nPlusOneThreshold", 2);
+      refresh();
+      for(int i = 0; i < 1005; i++)
+      {
+         try(var scope = UnitsOfWork.begin(UnitOfWorkKind.PROCESS, "job_" + i))
+         {
+            synthetic("select ?", 1, true, scope.getUnitOfWork());
+            synthetic("select ?", 1, true, scope.getUnitOfWork());
+         }
+      }
+      flusher.flushNow();
+      assertThat(rows("db_perf_attribution")).hasSize(1005);
+      assertThat(rows("db_perf_repeat_pattern")).hasSize(1005);
+      assertThat(CountingInsert.inputs).extracting(InsertInput::getTableName).containsExactlyInAnyOrder("db_perf_attribution", "db_perf_repeat_pattern");
+      assertThat(CountingInsert.inputs).allSatisfy(input -> assertThat(input.getRecords()).hasSize(1005));
+   }
+
+
+   /*******************************************************************************
+    ** An insert failure drops the window and never poisons a later good window.
+    *******************************************************************************/
+   @Test
+   void failedWindowIsDropped() throws Exception
+   {
+      RDBMSTableBackendDetails details = (RDBMSTableBackendDetails) runtime.getInstance().getTable("db_perf_attribution").getBackendDetails();
+      query(1);
+      details.setTableName("missing_attribution_table");
+      flusher.flushNow();
+      details.setTableName("db_perf_attribution");
+      flusher.flushNow();
+      assertThat(rows("db_perf_attribution")).isEmpty();
+      query(1);
+      flusher.flushNow();
+      assertThat(rows("db_perf_attribution")).singleElement().satisfies(row -> assertThat(row.getValueLong("statements")).isEqualTo(1));
+   }
+
+
+   /*******************************************************************************
+    ** Cold-cache overflow remains bounded and later hits keep the right shape.
+    *******************************************************************************/
+   @Test
+   void normalizationCacheOverflow()
+   {
+      for(int i = 0; i < 5010; i++)
+      {
+         synthetic("select column_" + i + " from business", 1, true, null);
+      }
+      synthetic("select column_0 from business", 1, true, null);
+      var window = listener.drain(Instant.now());
+      assertThat(window.attribution()).hasSize(5010);
+      String fingerprint = SqlNormalizer.fingerprint(SqlNormalizer.normalize("select column_0 from business"));
+      assertThat(window.attribution().entrySet().stream().filter(entry -> entry.getKey().fingerprint().equals(fingerprint)).findFirst().orElseThrow().getValue().statements.sum()).isEqualTo(2);
+   }
+
+
+   /*******************************************************************************
+    ** The daemon executes on its own system context and honors a short interval.
+    *******************************************************************************/
+   @Test
+   void scheduledFlush() throws Exception
+   {
+      settings.withValue("attributionFlushSeconds", 1);
+      refresh();
+      flusher.start();
+      flusher.start();
+      query(1);
+      long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+      while(rows("db_perf_attribution").isEmpty() && System.nanoTime() < deadline)
+      {
+         Thread.sleep(25);
+      }
+      assertThat(rows("db_perf_attribution")).hasSize(1);
+      assertThat(QContext.getQInstance()).isSameAs(runtime.getInstance());
+      flusher.close();
+      flusher.start();
+      assertThat(StatementExecutionListeners.isActive()).isFalse();
+   }
+
+
+   /*******************************************************************************
+    ** Deliver a statement with exact counter values.
+    *******************************************************************************/
+   private void synthetic(String sql, long nanos, boolean succeeded, UnitOfWork unit)
+   {
+      listener.onStatement(new StatementExecution(runtime.getConfig().getBackendName(), "business", StatementOperation.QUERY, sql, nanos, 1, succeeded, unit, null));
+   }
+
+
+   /*******************************************************************************
+    ** Execute real RDBMS-backed QQQ queries.
+    *******************************************************************************/
+   private void query(int count) throws Exception
+   {
+      for(int i = 0; i < count; i++)
+      {
+         new QueryAction().execute(new QueryInput("business").withFilter(new QQueryFilter()));
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Persist administrator settings, then explicitly refresh the cached copy.
+    *******************************************************************************/
+   private void refresh() throws Exception
+   {
+      DbPerfStore.update("db_perf_settings", List.of(settings));
+      flusher.refreshSettings();
+   }
+
+
+   /*******************************************************************************
+    ** Read persisted evidence with a concise test failure on action errors.
+    *******************************************************************************/
+   private List<QRecord> rows(String table)
+   {
+      try
+      {
+         return (DbPerfStore.query(table, new QQueryFilter()));
+      }
+      catch(Exception e)
+      {
+         throw (new AssertionError(e));
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Find one named aggregate.
+    *******************************************************************************/
+   private static QRecord named(List<QRecord> rows, String name)
+   {
+      return (rows.stream().filter(row -> name.equals(row.getValueString("unitName"))).findFirst().orElseThrow());
+   }
+
+   /*******************************************************************************
+    ** Observe whole QQQ actions while preserving the production write guard.
+    *******************************************************************************/
+   public static class CountingInsert extends DbPerfWriteGuard
+   {
+      private static final List<InsertInput> inputs = new ArrayList<>();
+
+
+      /*******************************************************************************
+       ** Capture each action's original input before the backend pages its SQL.
+       *******************************************************************************/
+      @Override
+      public List<QRecord> preInsert(InsertInput input, List<QRecord> records, boolean preview) throws QException
+      {
+         inputs.add(input);
+         return (super.preInsert(input, records, preview));
+      }
+   }
+}