refactor(BA-6927): rename usage_bucket_entries.amount to resource_usage

changes/12965.fix.md

@@ -0,0 +1 @@
+Fix user, project and domain usage buckets being over-reported by the number of concurrently running kernels, and rebuild the affected buckets from the recorded per-kernel usage.

changes/12968.misc.md

@@ -0,0 +1 @@
+Rename the internal `usage_bucket_entries.amount` column to `resource_usage`, matching the name the parent bucket tables already use for the same quantity.

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

@@ -417,15 +417,18 @@ class DomainUsageBucketKey:
 
 @dataclass
 class BucketDelta:
-    """Separated resource amount and duration for a usage bucket.
+    """Accumulated resource usage 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_usage`` is in resource-seconds and must be accumulated as
+    ``sum(amount_k * duration_k)``.  Summing amounts and durations separately
+    and multiplying afterwards gives a cross product inflated by the number of
+    slices in the bucket.
+
+    ``duration_seconds`` is for reporting only; both fields are sums, so they
+    are never multiplied with each other.
     """
 
-    slots: ResourceSlot = field(default_factory=ResourceSlot)
+    resource_usage: ResourceSlot = field(default_factory=ResourceSlot)
     duration_seconds: int = 0
 
 

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

@@ -0,0 +1,185 @@
+"""rebuild inflated usage buckets
+
+The aggregator summed amounts and durations separately and multiplied them
+afterwards, so both the JSONB ``resource_usage`` mirror and the normalized
+entries hold a cross product.  Neither can be corrected in place, so both are
+rebuilt from ``kernel_usage_records``, which was never affected.
+
+``usage_bucket_entries.amount`` also widens, because a domain-level daily mem
+bucket can exceed the previous 24-digit ceiling.
+
+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",
+        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:
+    # Keeps the rebuilt values; only the widened precision is reverted.
+    op.alter_column(
+        "usage_bucket_entries",
+        "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 because retention purges by ``period_end``
+    and may have truncated it.  Buckets outside the range keep their inflated
+    values rather than being zeroed: they are unrecoverable, and zeroing them
+    would destroy the only usage history left.
+    """
+    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
+        )
+    """
+
+    # capacity is refilled by the next observation tick, so dropping it is safe.
+    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, amount, 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 amount = EXCLUDED.amount
+            """
+        ),
+        {**params, "bucket_type": bucket_type},
+    )
+
+    # Buckets 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/alembic/versions/e7d3b95a1f26_rename_usage_bucket_entry_amount.py

@@ -0,0 +1,45 @@
+"""rename usage_bucket_entries.amount to resource_usage
+
+The column holds resource-seconds, not a resource amount.  Under the old name
+``amount * duration_seconds`` reads like the quantity a caller wants, when both
+columns are already sums and their product is a cross product.
+
+``resource_usage`` is the name the three parent bucket tables and
+``kernel_usage_records`` already use for this quantity, so the table now speaks
+the same vocabulary as the rest of the schema.
+
+Revision ID: e7d3b95a1f26
+Revises: c4a91d7e05b2
+Create Date: 2026-07-20 00:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "e7d3b95a1f26"
+down_revision = "c4a91d7e05b2"
+# Part of: NEXT_RELEASE_VERSION
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.alter_column(
+        "usage_bucket_entries",
+        "amount",
+        new_column_name="resource_usage",
+        existing_type=sa.Numeric(precision=32, scale=6),
+        existing_nullable=False,
+    )
+
+
+def downgrade() -> None:
+    op.alter_column(
+        "usage_bucket_entries",
+        "resource_usage",
+        new_column_name="amount",
+        existing_type=sa.Numeric(precision=32, scale=6),
+        existing_nullable=False,
+    )

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

@@ -442,10 +442,10 @@ 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_usage`` holds ``sum(amount_k * duration_k)`` resource-seconds and
+    is what the read paths sum, matching the column of the same name on the three
+    parent bucket tables.  ``duration_seconds`` is for reporting only; both
+    columns are sums, so they are never multiplied together.
 
     One entry per (bucket_id, slot_name). ``bucket_type`` is a discriminator
     indicating which parent table (domain/project/user_usage_buckets) owns
@@ -457,8 +457,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_usage: Mapped[Decimal] = mapped_column(
+        "resource_usage", 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_usage,
             )
             .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_usage,
             )
             .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_usage,
             )
             .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_usage))
 
         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_usage)
             )
 
         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_usage)
             )
 
         return RawUsageBucketsByLevel(