…ck (#705)

* test(inkless:retention): add enforce commit-blocking benchmark

Measures commit latency while enforce_retention_v2 runs on the same
partition, as an A/B harness for the lock-decoupling change. Two shapes:
single-partition (commit and enforce target the same partition, isolating
the scan-under-lock effect) and multi-partition (enforce runs over N
partitions while a commit targets the first, isolating lock accumulation
across partitions in one transaction). Tagged @tag("benchmark") so it is
excluded from CI and run on demand via the benchmarkTest task.

* fix(inkless:retention): decouple enforce_retention_v2 from the log lock

enforce_retention_v2 took logs FOR UPDATE before its O(depth) retention
boundary scan and looped every request in one transaction, so a
concurrent commit_file on a locked partition waited for the whole
enforcement pass. In production the commit-latency tail tracked the
enforce latency (commit max ~ enforce max).

Restructure enforcement so the lock is held only for the short delete:

- The retention boundary is now computed in a read-only step with no FOR
  UPDATE. byte_size is read inside that scan statement (a selected_log
  CTE) so the reverse-aggregate stays consistent with the batches rows
  under READ COMMITTED. The lock is taken only afterwards, for the delete.
- EnforceRetentionJob runs one transaction per request instead of wrapping
  all requests in one. A plpgsql function cannot release a row lock before
  it returns, so per-request transactions are what actually free the lock
  between partitions rather than accumulating it until the end. now is
  computed once per cycle and shared. Each transaction holds at most one
  partition lock, so enforce can no longer self-deadlock and the loop's
  ordering is no longer needed for deadlock avoidance on this path.
- Deleted batches/bytes are recounted under the lock so the reported
  counts stay accurate when a concurrent delete advances log_start between
  the unlocked scan and the locked delete.
- The max_batches_per_request cap now probes for the (cap+1)-th deletable
  batch with LIMIT/OFFSET instead of COUNT(*) over all deletable rows,
  keeping the locked section O(delete-size) rather than O(depth) when a
  small cap meets a large deletable set.
- The boundary scan is skipped entirely when the oldest retained batch
  provably survives every enabled policy, using logs.byte_size and
  logs.earliest_batch_timestamp. A NULL earliest_batch_timestamp means
  "unknown", so the short-circuit does not fire and the scan runs.

Computing the boundary unlocked is safe: retention deletes at the tail
while commit appends at the head, and delete_records_v1 re-clamps to the
current log_start under its own lock, so a concurrent commit or delete
cannot cause over-deletion. Partial progress across partitions is fine;
retention is idempotent and re-runs next cycle.

Adds PG concurrency tests for the interleavings this enables: log deleted
between scan and delete returns unknown_topic_or_partition; log_start
advanced between scan and delete returns the recounted deletion; and a
NULL earliest_batch_timestamp falls back to the scan instead of
short-circuiting.

Regenerates jOOQ sources for the schema-version stamp bump introduced by
the migration.

* feat(inkless:retention): mirror earliest_batch_timestamp to in-memory control plane

Mirror logs.earliest_batch_timestamp into InMemoryControlPlane for parity
with the Postgres control plane, so the enforce_retention short-circuit
behaves identically across both backends.

- LogInfo.earliestBatchTimestamp (nullable; null == "unknown, must scan",
  matching the SQL NULL sentinel), maintained under the same rules as V20:
  commit populates only when unset, delete/prune/enforce recompute from the
  new oldest batch when the oldest changed (null when the log empties).
- enforceRetention gains the same short-circuit: when the oldest retained
  batch survives every enabled policy, skip the O(depth) selection scans.
  It skips only the scans and falls through to the unchanged tail, so the
  reported log_start_offset and deletion counts are identical (pure no-op
  result-wise; the in-memory backend has no O(depth) scan to optimize, this
  is parity only).
- InMemoryControlPlaneEarliestBatchTimestampTest mirrors the Postgres
  LogsEarliestBatchTimestampTest across commit/delete/prune/enforce. The
  field is intentionally not on the ControlPlane interface, so it is read
  via reflection.

* fix(inkless:retention): stop in-memory enforceRetention moving log_start backward

InMemoryControlPlane.enforceRetention recomputed log_start_offset from the
oldest batch's base offset on every call, even when it deleted nothing. If a
prior deleteRecords had advanced log_start into the middle of a still-retained
batch (e.g. offset 5 inside batch [0,9]), a later enforce that deleted nothing
would reset log_start back to that batch's base (0). That diverges from
delete_records_v1 (Postgres) and re-exposes reads for offsets that should be
below the log start.

Only touch log_start (and the mirrored earliest_batch_timestamp) when a
deletion actually occurred, and make the recompute forward-only. Adds a parity
test in AbstractControlPlaneTest.Retention that runs against both backends:
after deleteRecords advances log_start mid-batch, an enforce that deletes
nothing must leave log_start unchanged.

…ck (#705)

* test(inkless:retention): add enforce commit-blocking benchmark

Measures commit latency while enforce_retention_v2 runs on the same
partition, as an A/B harness for the lock-decoupling change. Two shapes:
single-partition (commit and enforce target the same partition, isolating
the scan-under-lock effect) and multi-partition (enforce runs over N
partitions while a commit targets the first, isolating lock accumulation
across partitions in one transaction). Tagged @tag("benchmark") so it is
excluded from CI and run on demand via the benchmarkTest task.

* fix(inkless:retention): decouple enforce_retention_v2 from the log lock

enforce_retention_v2 took logs FOR UPDATE before its O(depth) retention
boundary scan and looped every request in one transaction, so a
concurrent commit_file on a locked partition waited for the whole
enforcement pass. In production the commit-latency tail tracked the
enforce latency (commit max ~ enforce max).

Restructure enforcement so the lock is held only for the short delete:

- The retention boundary is now computed in a read-only step with no FOR
  UPDATE. byte_size is read inside that scan statement (a selected_log
  CTE) so the reverse-aggregate stays consistent with the batches rows
  under READ COMMITTED. The lock is taken only afterwards, for the delete.
- EnforceRetentionJob runs one transaction per request instead of wrapping
  all requests in one. A plpgsql function cannot release a row lock before
  it returns, so per-request transactions are what actually free the lock
  between partitions rather than accumulating it until the end. now is
  computed once per cycle and shared. Each transaction holds at most one
  partition lock, so enforce can no longer self-deadlock and the loop's
  ordering is no longer needed for deadlock avoidance on this path.
- Deleted batches/bytes are recounted under the lock so the reported
  counts stay accurate when a concurrent delete advances log_start between
  the unlocked scan and the locked delete.
- The max_batches_per_request cap now probes for the (cap+1)-th deletable
  batch with LIMIT/OFFSET instead of COUNT(*) over all deletable rows,
  keeping the locked section O(delete-size) rather than O(depth) when a
  small cap meets a large deletable set.
- The boundary scan is skipped entirely when the oldest retained batch
  provably survives every enabled policy, using logs.byte_size and
  logs.earliest_batch_timestamp. A NULL earliest_batch_timestamp means
  "unknown", so the short-circuit does not fire and the scan runs.

Computing the boundary unlocked is safe: retention deletes at the tail
while commit appends at the head, and delete_records_v1 re-clamps to the
current log_start under its own lock, so a concurrent commit or delete
cannot cause over-deletion. Partial progress across partitions is fine;
retention is idempotent and re-runs next cycle.

Adds PG concurrency tests for the interleavings this enables: log deleted
between scan and delete returns unknown_topic_or_partition; log_start
advanced between scan and delete returns the recounted deletion; and a
NULL earliest_batch_timestamp falls back to the scan instead of
short-circuiting.

Regenerates jOOQ sources for the schema-version stamp bump introduced by
the migration.

* feat(inkless:retention): mirror earliest_batch_timestamp to in-memory control plane

Mirror logs.earliest_batch_timestamp into InMemoryControlPlane for parity
with the Postgres control plane, so the enforce_retention short-circuit
behaves identically across both backends.

- LogInfo.earliestBatchTimestamp (nullable; null == "unknown, must scan",
  matching the SQL NULL sentinel), maintained under the same rules as V20:
  commit populates only when unset, delete/prune/enforce recompute from the
  new oldest batch when the oldest changed (null when the log empties).
- enforceRetention gains the same short-circuit: when the oldest retained
  batch survives every enabled policy, skip the O(depth) selection scans.
  It skips only the scans and falls through to the unchanged tail, so the
  reported log_start_offset and deletion counts are identical (pure no-op
  result-wise; the in-memory backend has no O(depth) scan to optimize, this
  is parity only).
- InMemoryControlPlaneEarliestBatchTimestampTest mirrors the Postgres
  LogsEarliestBatchTimestampTest across commit/delete/prune/enforce. The
  field is intentionally not on the ControlPlane interface, so it is read
  via reflection.

* fix(inkless:retention): stop in-memory enforceRetention moving log_start backward

InMemoryControlPlane.enforceRetention recomputed log_start_offset from the
oldest batch's base offset on every call, even when it deleted nothing. If a
prior deleteRecords had advanced log_start into the middle of a still-retained
batch (e.g. offset 5 inside batch [0,9]), a later enforce that deleted nothing
would reset log_start back to that batch's base (0). That diverges from
delete_records_v1 (Postgres) and re-exposes reads for offsets that should be
below the log start.

Only touch log_start (and the mirrored earliest_batch_timestamp) when a
deletion actually occurred, and make the recompute forward-only. Adds a parity
test in AbstractControlPlaneTest.Retention that runs against both backends:
after deleteRecords advances log_start mid-batch, an enforce that deletes
nothing must leave log_start unchanged.

…O(cap)

The enforce_retention_v2 boundary scan reverse-aggregates byte_size over every
deletable batch, O(partition depth). PR #705 moved it out of the log lock so it
no longer blocks commit, but the scan itself stayed O(depth): on a bulk drain
(retention lowered so almost the whole partition is deletable) a single scan on
a multi-million-batch partition runs for seconds-to-minutes and exceeds the
enforcer's socket.timeout.ms (default 5s), aborting the call and orphaning the
still-running backend so the partition never drains.

V22 caps the scan to the oldest (max_batches_per_request + 1) batches. The
boundary can never be deeper than the cap because the delete is clamped there
regardless; the +1 distinguishes an in-window boundary (delete up to it) from a
deeper one that falls through to high_watermark and is re-clamped by the
existing under-lock cap-probe. Each pass is O(cap), index-only on
batches_by_last_offset_covering_idx, and a deep partition drains over
successive enforcement cycles. max_batches_per_request = 0 keeps the original
unbounded full scan via LIMIT NULL. Correctness is unchanged from V21 (deletes
oldest-first, so a snapshot-stale boundary can only under-delete).

Flip retention.enforcement.max.batches.per.request default 0 -> 1000 so the
bound is on out of the box; it must stay above the per-interval expiry rate or
retention falls behind.

Benchmark (EnforceRetentionCommitBlockingBenchmarkTest, now cap-overridable via
-Dinkless.benchmark.maxBatchesPerEnforce). enforce ms vs depth at 200k/400k/800k
batches: unbounded 359/729/1465 ms (linear O(depth)); bounded cap=1 21/21/29 ms
(flat). A cap sweep shows per-pass cost is flat from cap 1 to 1000, so 1000 is
the sweet spot: short passes at 10x the drain throughput of 100.

[KC-354]

…O(cap)

The enforce_retention_v2 boundary scan reverse-aggregates byte_size over every
deletable batch, O(partition depth). PR #705 moved it out of the log lock so it
no longer blocks commit, but the scan itself stayed O(depth): on a bulk drain
(retention lowered so almost the whole partition is deletable) a single scan on
a multi-million-batch partition runs for seconds-to-minutes and exceeds the
enforcer's socket.timeout.ms (default 5s), aborting the call and orphaning the
still-running backend so the partition never drains.

V22 caps the scan to the oldest (max_batches_per_request + 1) batches. The
boundary can never be deeper than the cap because the delete is clamped there
regardless; the +1 distinguishes an in-window boundary (delete up to it) from a
deeper one that falls through to high_watermark and is re-clamped by the
existing under-lock cap-probe. Each pass is O(cap), index-only on
batches_by_last_offset_covering_idx, and a deep partition drains over
successive enforcement cycles. max_batches_per_request = 0 keeps the original
unbounded full scan via LIMIT NULL. Correctness is unchanged from V21 (deletes
oldest-first, so a snapshot-stale boundary can only under-delete).

Flip retention.enforcement.max.batches.per.request default 0 -> 1000 so the
bound is on out of the box; it must stay above the per-interval expiry rate or
retention falls behind.

Benchmark (EnforceRetentionCommitBlockingBenchmarkTest, now cap-overridable via
-Dinkless.benchmark.maxBatchesPerEnforce). enforce ms vs depth at 200k/400k/800k
batches: unbounded 359/729/1465 ms (linear O(depth)); bounded cap=1 21/21/29 ms
(flat). A cap sweep shows per-pass cost is flat from cap 1 to 1000, so 1000 is
the sweet spot: short passes at 10x the drain throughput of 100.

[KC-354]

…O(cap) (#722)

The enforce_retention_v2 boundary scan reverse-aggregates byte_size over every
deletable batch, O(partition depth). PR #705 moved it out of the log lock so it
no longer blocks commit, but the scan itself stayed O(depth): on a bulk drain
(retention lowered so almost the whole partition is deletable) a single scan on
a multi-million-batch partition runs for seconds-to-minutes and exceeds the
enforcer's socket.timeout.ms (default 5s), aborting the call and orphaning the
still-running backend so the partition never drains.

V22 caps the scan to the oldest (max_batches_per_request + 1) batches. The
boundary can never be deeper than the cap because the delete is clamped there
regardless; the +1 distinguishes an in-window boundary (delete up to it) from a
deeper one that falls through to high_watermark and is re-clamped by the
existing under-lock cap-probe. Each pass is O(cap), index-only on
batches_by_last_offset_covering_idx, and a deep partition drains over
successive enforcement cycles. max_batches_per_request = 0 keeps the original
unbounded full scan via LIMIT NULL. Correctness is unchanged from V21 (deletes
oldest-first, so a snapshot-stale boundary can only under-delete).

Flip retention.enforcement.max.batches.per.request default 0 -> 1000 so the
bound is on out of the box; it must stay above the per-interval expiry rate or
retention falls behind.

Benchmark (EnforceRetentionCommitBlockingBenchmarkTest, now cap-overridable via
-Dinkless.benchmark.maxBatchesPerEnforce). enforce ms vs depth at 200k/400k/800k
batches: unbounded 359/729/1465 ms (linear O(depth)); bounded cap=1 21/21/29 ms
(flat). A cap sweep shows per-pass cost is flat from cap 1 to 1000, so 1000 is
the sweet spot: short passes at 10x the drain throughput of 100.

[KC-354]

EnforceRetentionJob runs one transaction per partition (post-#705), yet
EnforceRetentionQueryTime wrapped the whole batch -- duplicating the broker-side
RetentionEnforcementTotalTime, which already measures the whole enforce wave.

Record EnforceRetentionQueryTime/QueryRate per enforce_retention_v2 call (one per
partition) instead, matching the other *QueryTime metrics (one DB operation each)
and yielding per-partition enforce latency without a new metric. The whole-wave
total stays RetentionEnforcementTotalTime on the broker.

Note: this redefines two existing metrics -- EnforceRetentionQueryTime from
per-wave to per-partition, and EnforceRetentionQueryRate from wave count to
partitions-enforced count (wave count remains RetentionEnforcementRate).

EnforceRetentionJob runs one transaction per partition (post-#705), yet
EnforceRetentionQueryTime wrapped the whole batch -- duplicating the broker-side
RetentionEnforcementTotalTime, which already measures the whole enforce wave.

Record EnforceRetentionQueryTime/QueryRate per enforce_retention_v2 call (one per
partition) instead, matching the other *QueryTime metrics (one DB operation each)
and yielding per-partition enforce latency without a new metric. The whole-wave
total stays RetentionEnforcementTotalTime on the broker.

Note: this redefines two existing metrics -- EnforceRetentionQueryTime from
per-wave to per-partition, and EnforceRetentionQueryRate from wave count to
partitions-enforced count (wave count remains RetentionEnforcementRate).

…#725)

* feat(inkless:retention): surface enforcement schedule lag

Add a RetentionEnforcementScheduleLagMs gauge (0 when on schedule; a sustained
positive value means enforcement is falling behind its per-partition cadence),
fed by the scheduler's oldest past-due queue entry read under a lock so the
metrics/JMX thread and the enforcement thread do not race the PriorityQueue.

* feat(inkless:retention): meter enforce latency per partition

EnforceRetentionJob runs one transaction per partition (post-#705), yet
EnforceRetentionQueryTime wrapped the whole batch -- duplicating the broker-side
RetentionEnforcementTotalTime, which already measures the whole enforce wave.

Record EnforceRetentionQueryTime/QueryRate per enforce_retention_v2 call (one per
partition) instead, matching the other *QueryTime metrics (one DB operation each)
and yielding per-partition enforce latency without a new metric. The whole-wave
total stays RetentionEnforcementTotalTime on the broker.

Note: this redefines two existing metrics -- EnforceRetentionQueryTime from
per-wave to per-partition, and EnforceRetentionQueryRate from wave count to
partitions-enforced count (wave count remains RetentionEnforcementRate).

* feat(inkless:retention): make enforcement and file-cleaner logs intent/result pairs

Retention enforcer: promote the per-cycle summary from DEBUG to INFO and add a
matching intent line before the (potentially blocking) enforce call, so a stuck
enforce shows as intent-without-completion:
  "Enforcing retention for {n} partitions"
  "Enforced retention for {n} partitions in {ms} ms: {b} batches, {y} bytes deleted"
One pair per non-empty cycle; empty cycles stay silent (no per-tick heartbeat).
Also fix a latent partition misattribution: the per-partition log lines indexed
readyPartitions while responses align with the delete-policy-filtered requests, so a
filtered-out partition could be mislabeled -- keep an enforcedPartitions list aligned
1:1 with requests/responses instead.

File cleaner: replace the per-run "Running file cleaner at {now}" heartbeat
(redundant with the framework timestamp and the outcome line) with a work-gated
intent/result pair:
  "Running file cleaner: deleting {n} of {m} marked files"  (m-n still within grace)
  "File cleaner deleted {n} files in {ms} ms"
Partition/size counts are not available at this layer (FileToDelete carries only the
object key; batches are already deleted by cleanup time).

…O(cap) (#722)

The enforce_retention_v2 boundary scan reverse-aggregates byte_size over every
deletable batch, O(partition depth). PR #705 moved it out of the log lock so it
no longer blocks commit, but the scan itself stayed O(depth): on a bulk drain
(retention lowered so almost the whole partition is deletable) a single scan on
a multi-million-batch partition runs for seconds-to-minutes and exceeds the
enforcer's socket.timeout.ms (default 5s), aborting the call and orphaning the
still-running backend so the partition never drains.

V22 caps the scan to the oldest (max_batches_per_request + 1) batches. The
boundary can never be deeper than the cap because the delete is clamped there
regardless; the +1 distinguishes an in-window boundary (delete up to it) from a
deeper one that falls through to high_watermark and is re-clamped by the
existing under-lock cap-probe. Each pass is O(cap), index-only on
batches_by_last_offset_covering_idx, and a deep partition drains over
successive enforcement cycles. max_batches_per_request = 0 keeps the original
unbounded full scan via LIMIT NULL. Correctness is unchanged from V21 (deletes
oldest-first, so a snapshot-stale boundary can only under-delete).

Flip retention.enforcement.max.batches.per.request default 0 -> 1000 so the
bound is on out of the box; it must stay above the per-interval expiry rate or
retention falls behind.

Benchmark (EnforceRetentionCommitBlockingBenchmarkTest, now cap-overridable via
-Dinkless.benchmark.maxBatchesPerEnforce). enforce ms vs depth at 200k/400k/800k
batches: unbounded 359/729/1465 ms (linear O(depth)); bounded cap=1 21/21/29 ms
(flat). A cap sweep shows per-pass cost is flat from cap 1 to 1000, so 1000 is
the sweet spot: short passes at 10x the drain throughput of 100.

[KC-354]

…#725)

* feat(inkless:retention): surface enforcement schedule lag

Add a RetentionEnforcementScheduleLagMs gauge (0 when on schedule; a sustained
positive value means enforcement is falling behind its per-partition cadence),
fed by the scheduler's oldest past-due queue entry read under a lock so the
metrics/JMX thread and the enforcement thread do not race the PriorityQueue.

* feat(inkless:retention): meter enforce latency per partition

EnforceRetentionJob runs one transaction per partition (post-#705), yet
EnforceRetentionQueryTime wrapped the whole batch -- duplicating the broker-side
RetentionEnforcementTotalTime, which already measures the whole enforce wave.

Record EnforceRetentionQueryTime/QueryRate per enforce_retention_v2 call (one per
partition) instead, matching the other *QueryTime metrics (one DB operation each)
and yielding per-partition enforce latency without a new metric. The whole-wave
total stays RetentionEnforcementTotalTime on the broker.

Note: this redefines two existing metrics -- EnforceRetentionQueryTime from
per-wave to per-partition, and EnforceRetentionQueryRate from wave count to
partitions-enforced count (wave count remains RetentionEnforcementRate).

* feat(inkless:retention): make enforcement and file-cleaner logs intent/result pairs

Retention enforcer: promote the per-cycle summary from DEBUG to INFO and add a
matching intent line before the (potentially blocking) enforce call, so a stuck
enforce shows as intent-without-completion:
  "Enforcing retention for {n} partitions"
  "Enforced retention for {n} partitions in {ms} ms: {b} batches, {y} bytes deleted"
One pair per non-empty cycle; empty cycles stay silent (no per-tick heartbeat).
Also fix a latent partition misattribution: the per-partition log lines indexed
readyPartitions while responses align with the delete-policy-filtered requests, so a
filtered-out partition could be mislabeled -- keep an enforcedPartitions list aligned
1:1 with requests/responses instead.

File cleaner: replace the per-run "Running file cleaner at {now}" heartbeat
(redundant with the framework timestamp and the outcome line) with a work-gated
intent/result pair:
  "Running file cleaner: deleting {n} of {m} marked files"  (m-n still within grace)
  "File cleaner deleted {n} files in {ms} ms"
Partition/size counts are not available at this layer (FileToDelete carries only the
object key; batches are already deleted by cleanup time).

…O(cap) (#722)

The enforce_retention_v2 boundary scan reverse-aggregates byte_size over every
deletable batch, O(partition depth). PR #705 moved it out of the log lock so it
no longer blocks commit, but the scan itself stayed O(depth): on a bulk drain
(retention lowered so almost the whole partition is deletable) a single scan on
a multi-million-batch partition runs for seconds-to-minutes and exceeds the
enforcer's socket.timeout.ms (default 5s), aborting the call and orphaning the
still-running backend so the partition never drains.

V22 caps the scan to the oldest (max_batches_per_request + 1) batches. The
boundary can never be deeper than the cap because the delete is clamped there
regardless; the +1 distinguishes an in-window boundary (delete up to it) from a
deeper one that falls through to high_watermark and is re-clamped by the
existing under-lock cap-probe. Each pass is O(cap), index-only on
batches_by_last_offset_covering_idx, and a deep partition drains over
successive enforcement cycles. max_batches_per_request = 0 keeps the original
unbounded full scan via LIMIT NULL. Correctness is unchanged from V21 (deletes
oldest-first, so a snapshot-stale boundary can only under-delete).

Flip retention.enforcement.max.batches.per.request default 0 -> 1000 so the
bound is on out of the box; it must stay above the per-interval expiry rate or
retention falls behind.

Benchmark (EnforceRetentionCommitBlockingBenchmarkTest, now cap-overridable via
-Dinkless.benchmark.maxBatchesPerEnforce). enforce ms vs depth at 200k/400k/800k
batches: unbounded 359/729/1465 ms (linear O(depth)); bounded cap=1 21/21/29 ms
(flat). A cap sweep shows per-pass cost is flat from cap 1 to 1000, so 1000 is
the sweet spot: short passes at 10x the drain throughput of 100.

[KC-354]

…#725)

* feat(inkless:retention): surface enforcement schedule lag

Add a RetentionEnforcementScheduleLagMs gauge (0 when on schedule; a sustained
positive value means enforcement is falling behind its per-partition cadence),
fed by the scheduler's oldest past-due queue entry read under a lock so the
metrics/JMX thread and the enforcement thread do not race the PriorityQueue.

* feat(inkless:retention): meter enforce latency per partition

EnforceRetentionJob runs one transaction per partition (post-#705), yet
EnforceRetentionQueryTime wrapped the whole batch -- duplicating the broker-side
RetentionEnforcementTotalTime, which already measures the whole enforce wave.

Record EnforceRetentionQueryTime/QueryRate per enforce_retention_v2 call (one per
partition) instead, matching the other *QueryTime metrics (one DB operation each)
and yielding per-partition enforce latency without a new metric. The whole-wave
total stays RetentionEnforcementTotalTime on the broker.

Note: this redefines two existing metrics -- EnforceRetentionQueryTime from
per-wave to per-partition, and EnforceRetentionQueryRate from wave count to
partitions-enforced count (wave count remains RetentionEnforcementRate).

* feat(inkless:retention): make enforcement and file-cleaner logs intent/result pairs

Retention enforcer: promote the per-cycle summary from DEBUG to INFO and add a
matching intent line before the (potentially blocking) enforce call, so a stuck
enforce shows as intent-without-completion:
  "Enforcing retention for {n} partitions"
  "Enforced retention for {n} partitions in {ms} ms: {b} batches, {y} bytes deleted"
One pair per non-empty cycle; empty cycles stay silent (no per-tick heartbeat).
Also fix a latent partition misattribution: the per-partition log lines indexed
readyPartitions while responses align with the delete-policy-filtered requests, so a
filtered-out partition could be mislabeled -- keep an enforcedPartitions list aligned
1:1 with requests/responses instead.

File cleaner: replace the per-run "Running file cleaner at {now}" heartbeat
(redundant with the framework timestamp and the outcome line) with a work-gated
intent/result pair:
  "Running file cleaner: deleting {n} of {m} marked files"  (m-n still within grace)
  "File cleaner deleted {n} files in {ms} ms"
Partition/size counts are not available at this layer (FileToDelete carries only the
object key; batches are already deleted by cleanup time).