fix(BA-6927): stop cross-multiplying usage bucket amounts and durations

FairShareAggregator accumulated the raw resource amounts and the slice
durations of a bucket separately and multiplied them afterwards, yielding
(sum amount_k) * (sum duration_k) instead of sum(amount_k * duration_k).
The cross product inflates a bucket by the number of kernel slices folded
into it, so a user running 4 GPU kernels saw 4x their real fGPU-seconds
and project/domain buckets were off by more, since they aggregate the
kernels of every user beneath them.

BucketDelta now carries resource-seconds directly, multiplied per slice
before accumulation, and the three repository call sites no longer
re-multiply by the accumulated duration.

usage_bucket_entries.amount is renamed to resource_seconds to match what
it holds; its precision widens to 32 digits because a domain-level daily
mem bucket can exceed the previous 1e18 ceiling. duration_seconds stays
as a reporting-only total and is never multiplied with resource_seconds.

Existing buckets cannot be corrected in place, in either the JSONB mirror
or the normalized entries, so the migration rebuilds both from
kernel_usage_records, which stores per-slice resource-seconds and was
never affected. The oldest retained day is excluded because retention
purges kernel records by period_end and may have truncated it; buckets
older than the kernel record retention window keep their inflated values
rather than being silently zeroed.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: jopemachine

src/ai/backend/manager/data/fair_share/types.py

@@ -417,15 +417,24 @@ class DomainUsageBucketKey:
 
 @dataclass
 class BucketDelta:
-    """Separated resource amount and duration for a usage bucket.
+    """Accumulated resource-seconds and observation duration for a usage bucket.
 
-    Stores raw resource amounts and duration separately instead of
-    pre-multiplied resource-seconds.  The product ``amount * duration_seconds``
-    is computed at SQL query time where PostgreSQL auto-extends NUMERIC precision,
-    eliminating overflow risk for large memory values.
+    ``resource_seconds`` is the time-weighted total ``sum(amount_k * duration_k)``
+    over every observation slice folded into this bucket, and is the value the
+    usage bucket tables store.
+
+    It MUST be accumulated as a sum of per-slice products.  Accumulating the raw
+    amounts and the durations separately and multiplying them afterwards yields
+    ``(sum amount_k) * (sum duration_k)``, a cross product that inflates the
+    result by the number of slices merged into the bucket -- i.e. by the number
+    of concurrently running kernels.
+
+    ``duration_seconds`` is the total observed duration (kernel-seconds, so
+    concurrent kernels each contribute their own slice length).  It is carried
+    for reporting only and is never multiplied with ``resource_seconds``.
     """
 
-    slots: ResourceSlot = field(default_factory=ResourceSlot)
+    resource_seconds: ResourceSlot = field(default_factory=ResourceSlot)
     duration_seconds: int = 0
 
 

src/ai/backend/manager/models/alembic/versions/c4a91d7e05b2_fix_usage_bucket_cross_product.py

@@ -0,0 +1,198 @@
+"""fix usage bucket cross-product inflation
+
+Renames ``usage_bucket_entries.amount`` to ``resource_seconds`` and rebuilds the
+corrupted aggregates from ``kernel_usage_records``.
+
+The aggregator used to accumulate the raw resource amounts and the slice
+durations into a bucket separately and multiply them afterwards, producing
+``(sum amount_k) * (sum duration_k)`` instead of ``sum(amount_k * duration_k)``.
+That cross product inflates every bucket by the number of kernel slices folded
+into it, so a user running 4 kernels saw 4x their real usage and project/domain
+buckets were off by even more.
+
+Both representations are affected: the JSONB ``resource_usage`` column on the
+three bucket tables and the normalized ``usage_bucket_entries`` rows.  Neither
+can be corrected in place, so both are rebuilt from ``kernel_usage_records``,
+which stores per-slice resource-seconds and was never affected.
+
+Revision ID: c4a91d7e05b2
+Revises: 3f9a1c7b2e04
+Create Date: 2026-07-20 00:00:00.000000
+
+"""
+
+from datetime import date, timedelta
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "c4a91d7e05b2"
+down_revision = "3f9a1c7b2e04"
+# Part of: NEXT_RELEASE_VERSION
+branch_labels = None
+depends_on = None
+
+
+# Bucket tables keyed by the columns that identify the owning entity.  Every key
+# column has the same name in kernel_usage_records, so the join is by name.
+_BUCKET_LEVELS: list[tuple[str, str, list[str]]] = [
+    ("user_usage_buckets", "user", ["user_uuid", "project_id", "resource_group"]),
+    ("project_usage_buckets", "project", ["project_id", "resource_group"]),
+    ("domain_usage_buckets", "domain", ["domain_name", "resource_group"]),
+]
+
+
+def upgrade() -> None:
+    op.alter_column(
+        "usage_bucket_entries",
+        "amount",
+        new_column_name="resource_seconds",
+        existing_type=sa.Numeric(precision=24, scale=6),
+        type_=sa.Numeric(precision=32, scale=6),
+        existing_nullable=False,
+    )
+
+    conn = op.get_bind()
+    covered = _covered_date_range(conn)
+    if covered is None:
+        # No usage records to rebuild from (fresh install, or everything purged).
+        return
+    rebuild_from, rebuild_to = covered
+    for table_name, bucket_type, key_columns in _BUCKET_LEVELS:
+        _rebuild_buckets(conn, table_name, bucket_type, key_columns, rebuild_from, rebuild_to)
+
+
+def downgrade() -> None:
+    # The rebuilt values are correct resource-seconds; reverting the rename keeps
+    # them, since the pre-fix column held a (differently scaled) usage figure too.
+    op.alter_column(
+        "usage_bucket_entries",
+        "resource_seconds",
+        new_column_name="amount",
+        existing_type=sa.Numeric(precision=32, scale=6),
+        type_=sa.Numeric(precision=24, scale=6),
+        existing_nullable=False,
+    )
+
+
+def _covered_date_range(conn: sa.engine.Connection) -> tuple[date, date] | None:
+    """Return the date range that kernel_usage_records can faithfully rebuild.
+
+    The oldest retained day is excluded: retention purges kernel usage records by
+    ``period_end``, so the boundary day may be partially purged and would rebuild
+    to an under-counted value.  Buckets outside this range keep their inflated
+    values rather than being silently zeroed -- they cannot be recovered, and
+    destroying them would lose the only usage history that remains.
+    """
+    row = conn.execute(
+        sa.text(
+            "SELECT min((period_start AT TIME ZONE 'UTC')::date) AS min_date, "
+            "       max((period_start AT TIME ZONE 'UTC')::date) AS max_date "
+            "FROM kernel_usage_records"
+        )
+    ).one()
+    if row.min_date is None or row.max_date is None:
+        return None
+    rebuild_from = row.min_date + timedelta(days=1)
+    if rebuild_from > row.max_date:
+        return None
+    return rebuild_from, row.max_date
+
+
+def _rebuild_buckets(
+    conn: sa.engine.Connection,
+    table_name: str,
+    bucket_type: str,
+    key_columns: list[str],
+    rebuild_from: date,
+    rebuild_to: date,
+) -> None:
+    """Recompute one bucket level's entries and JSONB from kernel_usage_records."""
+    key_list = ", ".join(key_columns)
+    join_on = " AND ".join(f"b.{col} = agg.{col}" for col in key_columns)
+    params = {"rebuild_from": rebuild_from, "rebuild_to": rebuild_to}
+
+    # Per-slot resource-seconds, summed over every kernel slice of the day.
+    agg_cte = f"""
+        WITH agg AS (
+            SELECT {key_list},
+                   (period_start AT TIME ZONE 'UTC')::date AS period_date,
+                   kv.key AS slot_name,
+                   SUM(kv.value::numeric) AS resource_seconds
+            FROM kernel_usage_records,
+                 LATERAL jsonb_each_text(resource_usage) AS kv
+            WHERE (period_start AT TIME ZONE 'UTC')::date
+                  BETWEEN :rebuild_from AND :rebuild_to
+            GROUP BY {key_list}, period_date, kv.key
+        )
+    """
+
+    # Drop the stale entries for the rebuilt window; capacity is refilled by the
+    # next fair share observation tick (update_bucket_entry_capacities).
+    conn.execute(
+        sa.text(
+            f"""
+            DELETE FROM usage_bucket_entries e
+            USING {table_name} b
+            WHERE e.bucket_id = b.id
+              AND e.bucket_type = :bucket_type
+              AND b.period_start BETWEEN :rebuild_from AND :rebuild_to
+            """
+        ),
+        {**params, "bucket_type": bucket_type},
+    )
+    conn.execute(
+        sa.text(
+            f"""
+            {agg_cte}
+            INSERT INTO usage_bucket_entries
+                (bucket_id, bucket_type, slot_name, resource_seconds, duration_seconds, capacity)
+            SELECT b.id, :bucket_type, agg.slot_name, agg.resource_seconds, 0, 0
+            FROM agg
+            JOIN {table_name} b
+              ON {join_on}
+             AND b.period_start = agg.period_date
+            ON CONFLICT (bucket_id, slot_name) DO UPDATE
+                SET resource_seconds = EXCLUDED.resource_seconds
+            """
+        ),
+        {**params, "bucket_type": bucket_type},
+    )
+
+    # Rebuild the JSONB mirror on the parent bucket rows.  Buckets in the window
+    # with no matching kernel records collapse to an empty slot map.
+    conn.execute(
+        sa.text(
+            f"""
+            {agg_cte},
+            per_bucket AS (
+                SELECT {key_list}, period_date,
+                       jsonb_object_agg(slot_name, resource_seconds) AS usage
+                FROM agg
+                GROUP BY {key_list}, period_date
+            )
+            UPDATE {table_name} b
+            SET resource_usage = COALESCE(agg.usage, '{{}}'::jsonb)
+            FROM per_bucket agg
+            WHERE {join_on}
+              AND b.period_start = agg.period_date
+              AND b.period_start BETWEEN :rebuild_from AND :rebuild_to
+            """
+        ),
+        params,
+    )
+    conn.execute(
+        sa.text(
+            f"""
+            UPDATE {table_name} b
+            SET resource_usage = '{{}}'::jsonb
+            WHERE b.period_start BETWEEN :rebuild_from AND :rebuild_to
+              AND NOT EXISTS (
+                  SELECT 1 FROM usage_bucket_entries e
+                  WHERE e.bucket_id = b.id AND e.bucket_type = :bucket_type
+              )
+            """
+        ),
+        {**params, "bucket_type": bucket_type},
+    )

src/ai/backend/manager/models/resource_usage_history/row.py

@@ -442,10 +442,11 @@ class UserUsageBucketRow(LifecycleTimestampsMixin, Base):  # type: ignore[misc]
 class UsageBucketEntryRow(Base):  # type: ignore[misc]
     """Per-slot normalized entry for usage bucket aggregation (Phase 3).
 
-    Stores amount and duration separately instead of pre-multiplied resource-seconds,
-    eliminating overflow risk for large memory values.
-    The product ``amount * duration_seconds`` is computed at SQL query time
-    where PostgreSQL auto-extends NUMERIC precision.
+    ``resource_seconds`` is the time-weighted total ``sum(amount_k * duration_k)``
+    over every observation slice folded into this bucket, and is what the read
+    paths sum.  ``duration_seconds`` is the total observed duration, kept for
+    reporting only -- the two columns are never multiplied together, since both
+    are already sums and their product would be a cross product.
 
     One entry per (bucket_id, slot_name). ``bucket_type`` is a discriminator
     indicating which parent table (domain/project/user_usage_buckets) owns
@@ -457,8 +458,8 @@ class UsageBucketEntryRow(Base):  # type: ignore[misc]
     bucket_id: Mapped[uuid.UUID] = mapped_column("bucket_id", GUID(), nullable=False)
     bucket_type: Mapped[str] = mapped_column("bucket_type", sa.String(length=16), nullable=False)
     slot_name: Mapped[str] = mapped_column("slot_name", sa.String(length=64), nullable=False)
-    amount: Mapped[Decimal] = mapped_column(
-        "amount", sa.Numeric(precision=24, scale=6), nullable=False
+    resource_seconds: Mapped[Decimal] = mapped_column(
+        "resource_seconds", sa.Numeric(precision=32, scale=6), nullable=False
     )
     duration_seconds: Mapped[int] = mapped_column("duration_seconds", sa.Integer(), nullable=False)
     capacity: Mapped[Decimal] = mapped_column(

src/ai/backend/manager/repositories/fair_share/db_source/db_source.py

@@ -1585,7 +1585,7 @@ async def _fetch_raw_usage_buckets(
                 UserUsageBucketRow.project_id,
                 UserUsageBucketRow.period_start,
                 ube.c.slot_name,
-                ube.c.amount,
+                ube.c.resource_seconds,
             )
             .select_from(
                 sa.join(
@@ -1612,7 +1612,7 @@ async def _fetch_raw_usage_buckets(
                 ProjectUsageBucketRow.project_id,
                 ProjectUsageBucketRow.period_start,
                 ube.c.slot_name,
-                ube.c.amount,
+                ube.c.resource_seconds,
             )
             .select_from(
                 sa.join(
@@ -1639,7 +1639,7 @@ async def _fetch_raw_usage_buckets(
                 DomainUsageBucketRow.domain_name,
                 DomainUsageBucketRow.period_start,
                 ube.c.slot_name,
-                ube.c.amount,
+                ube.c.resource_seconds,
             )
             .select_from(
                 sa.join(
@@ -1668,7 +1668,7 @@ async def _fetch_raw_usage_buckets(
                 user_buckets[key] = {}
             if row.period_start not in user_buckets[key]:
                 user_buckets[key][row.period_start] = ResourceSlot()
-            user_buckets[key][row.period_start][row.slot_name] = Decimal(str(row.amount))
+            user_buckets[key][row.period_start][row.slot_name] = Decimal(str(row.resource_seconds))
 
         project_buckets: dict[uuid.UUID, dict[date, ResourceSlot]] = {}
         for row in project_rows:
@@ -1677,7 +1677,7 @@ async def _fetch_raw_usage_buckets(
             if row.period_start not in project_buckets[row.project_id]:
                 project_buckets[row.project_id][row.period_start] = ResourceSlot()
             project_buckets[row.project_id][row.period_start][row.slot_name] = Decimal(
-                str(row.amount)
+                str(row.resource_seconds)
             )
 
         domain_buckets: dict[str, dict[date, ResourceSlot]] = {}
@@ -1687,7 +1687,7 @@ async def _fetch_raw_usage_buckets(
             if row.period_start not in domain_buckets[row.domain_name]:
                 domain_buckets[row.domain_name][row.period_start] = ResourceSlot()
             domain_buckets[row.domain_name][row.period_start][row.slot_name] = Decimal(
-                str(row.amount)
+                str(row.resource_seconds)
             )
 
         return RawUsageBucketsByLevel(