feat: replace credit balance cache with per-expiration buckets

Replaces the previous scalar cache + lazy-recompute design with a bucket
cache keyed by (address, expiration_date), maintained eagerly by the
credit_history writers under a sort-by-expiration policy. The API read
path is now a single SELECT with no writes, no FIFO walk, and no cache
invalidation logic.

Key changes:

- New credit_balances schema: (address, expiration_date, amount, last_update)
  with composite PK. The "no expiration" case uses a sentinel timestamptz
  (9999-12-31) so the column can stay NOT NULL.

- Writers (distribution, expense, transfer) update buckets atomically in
  the same transaction as the credit_history insert.
  Expenses drain the soonest-expiring non-expired bucket first under
  SELECT ... FOR UPDATE, serialising concurrent writers per address.

- Transfers compute recipient bucket expirations by capping each
  consumed source bucket at min(source_expiration, requested_expiration),
  preserving the existing re-transfer guard.

- New background task (CreditExpirationTask) sleeps until the next
  pending expiration timestamp and deletes expired buckets when it
  fires. Writers wake the task via a module-level asyncio.Event when
  they insert a bucket whose expiration may be sooner than the one
  currently being awaited. Expirations remain implicit (no synthetic
  credit_history rows).

- get_credit_balance / get_credit_balance_with_details are pure SELECTs:
  no FIFO recompute, no write-back. _apply_fifo_consumption and
  _calculate_credit_balance_fifo are removed.

- Repair (aleph.repair.repair_node) now also bootstraps and reconciles
  the bucket cache from credit_history, runs on every startup, and is
  idempotent. Replaces the previous one-shot script approach.

- Policy change: consumption now sorts by expiration_date (soonest
  first) rather than by message_timestamp (oldest first). This loses
  fewer credits to expiration when issuance order and expiration order
  diverge. FIFO scenario 1 in tests was updated accordingly; scenario 2
  remains unchanged since both policies produce the same result there.

The previous credit_balances rows are dropped by the migration; the
repair function repopulates them on the next startup from credit_history,
which is the source of truth.

Test plan covered:
- bucket helper unit tests (insert/increment, negative/under-funded
  consumption, sort-by-expiration drain order, sentinel handling)
- writer tests (distribution buckets, expense drain order, transfer
  cap, whitelisted sender, recipient inherits capped expiration)
- read path (sum non-expired buckets, with_details grouping, multi-
  address listing + count)
- expiration task (deletes expired, ignores sentinel, wakes on signal)
- repair (rebuilds buckets from history, idempotent)
- existing transfer / cap / expiration tests remain green under the new
  semantics
- removed test_cache_invalidation_on_credit_expiration (no cache to
  invalidate); replaced with a smaller test asserting that the read
  query filters expired buckets directly

Author: odesenfans

deployment/migrations/versions/0057_b1c2d3e4f5a6_replace_credit_balances_with_buckets.py

@@ -0,0 +1,68 @@
+"""Replace credit_balances scalar cache with per-expiration buckets
+
+Revision ID: b1c2d3e4f5a6
+Revises: 7e5a630e4b36
+Create Date: 2026-05-12
+
+The previous credit_balances table held a single FIFO-derived scalar per
+address, lazily recomputed on read by an O(N^2) Python walk over
+credit_history. This migration replaces it with a bucket cache keyed by
+(address, expiration_date), eagerly maintained by the credit_history
+writers under a sort-by-expiration policy. The "no expiration" case is
+encoded with a sentinel timestamptz of 9999-12-31T23:59:59Z so that
+expiration_date is NOT NULL and can sit in the composite primary key.
+
+The previous table is dropped rather than migrated; credit_history is the
+source of truth and aleph.repair.repair_credit_balances repopulates the
+new table from history on the next startup.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "b1c2d3e4f5a6"
+down_revision = "7e5a630e4b36"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.drop_table("credit_balances")
+    op.create_table(
+        "credit_balances",
+        sa.Column("address", sa.String(), nullable=False),
+        sa.Column("expiration_date", sa.TIMESTAMP(timezone=True), nullable=False),
+        sa.Column("amount", sa.BigInteger(), nullable=False),
+        sa.Column(
+            "last_update",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+        sa.PrimaryKeyConstraint(
+            "address", "expiration_date", name="credit_balances_pkey"
+        ),
+    )
+    op.create_index(
+        "credit_balances_expiration_date_idx",
+        "credit_balances",
+        ["expiration_date"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("credit_balances")
+    op.create_table(
+        "credit_balances",
+        sa.Column("address", sa.String(), nullable=False),
+        sa.Column("balance", sa.BigInteger(), nullable=False, server_default="0"),
+        sa.Column(
+            "last_update",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+        sa.PrimaryKeyConstraint("address", name="credit_balances_pkey"),
+    )
+    op.create_index("ix_credit_balances_address", "credit_balances", ["address"])

src/aleph/commands.py

@@ -36,6 +36,7 @@
 from aleph.repair import repair_node
 from aleph.services import p2p
 from aleph.services.cache.node_cache import NodeCache
+from aleph.services.credit_expiration import CreditExpirationTask
 from aleph.services.ipfs import IpfsService
 from aleph.services.keys import generate_keypair, save_keys
 from aleph.services.storage.fileystem_engine import FileSystemStorageEngine
@@ -224,6 +225,11 @@ async def main(args: List[str]) -> None:
         tasks.append(cron_job_task(config=config, cron_job=cron_job))
         LOGGER.debug("Initialized cron job task")
 
+        LOGGER.debug("Initializing credit expiration task")
+        credit_expiration = CreditExpirationTask(session_factory=session_factory)
+        tasks.append(credit_expiration.run())
+        LOGGER.debug("Initialized credit expiration task")
+
         LOGGER.debug("Running event loop")
         await asyncio.gather(*tasks)