8000 feat(inkless:delete): soft-delete diskless topics and purge in the background [KC-349] by viktorsomogyi · Pull Request #782 · aiven/inkless · GitHub
[go: up one dir, main page]
Skip to content

feat(inkless:delete): soft-delete diskless topics and purge in the background [KC-349] - #782

Merged
giuseppelillo merged 3 commits into
mainfrom
svv/diskless-async-delete
Sep 4, 2026
Merged

feat(inkless:delete): soft-delete diskless topics and purge in the background [KC-349]#782
giuseppelillo merged 3 commits into
mainfrom
svv/diskless-async-delete

Conversation

@viktorsomogyi
@viktorsomogyi viktorsomogyi commented Sep 2, 2026 •
edited
Loading
Copy link
Copy Markdown
Contributor

The old delete_topic_v1 deleted every batch for the topic on the controller event-handler thread. A large diskless topic made DELETE_TOPICS unbounded.

DELETE_TOPICS now only stamps logs.deleted_at. Live readers treat a stamped row as gone. Brokers run TopicPurger, which calls purge_deleted_logs_v1 with a cap. FileCleaner still deletes the objects
after file.cleaner.retention.period.ms.

Controller path

delete_topic_v1 takes the live log rows FOR UPDATE, then sets deleted_at. It does not delete batches, drop logs, or mark files. ReplicaManager schedules inkless-topic-purger at topic.purger.interval.ms (default 60s).

Purge

Empty deleted logs: drop producer_state, then the log. The same per-cycle cap applies, so a wide empty topic cannot take an unbounded number of row locks. Leftovers keep more_remain true.

Logs with batches: SKIP LOCKED, drain up to the remaining budget, then drop producer_state and the log when no batches remain. The leftover probe is EXISTS (... LIMIT 1), not COUNT(*).

Files: lock affected files by file_id before the emptiness check. Two brokers can drain different partitions of one WAL file. Without that lock, neither transaction sees the other's uncommitted deletes, both skip mark_file_to_delete_v1, and the object stays uploaded with zero batches.

more_remain is a cluster-wide EXISTS(deleted_at). TopicPurger treats a cycle as idle when it deleted, purged, and marked nothing, so a SKIP LOCKED miss does not spin. A cycle is saturated when batchesDeleted reaches the cap and more_remain is still true.

producer_state

The old delete_topic_v1 never removed producer_state, and there is no foreign key to logs. A purge regression would restore that leak. Purge deletes those rows when it drops a log. V28 also deletes orphans that have no log row, because the purger only walks soft-deleted logs.

Live readers

SQL functions that join or lock logs gain deleted_at IS NULL. find_batches_v2 keeps V27's request-level l_global_bytes = 0 grant. GetLogInfo, GetProducerState, and GetCrossTierLogStart apply the same filter in Java. InMemoryControlPlane mirrors this with liveLog.

Config and metrics

topic.purger.interval.ms defaults to 60000. topic.purger.max.batches.per.cycle defaults to 20000. The same bound caps empty-log drops. 0 means unbounded. With those defaults each broker drains at most 20000 batches per minute.

TopicPurger metrics: cycle time and rate, batches/logs/files, errors, TopicPurgerCycleSaturatedRate, and TopicPurgerWorkRemain (1 when the last successful cycle left work).

FEATURES.md: DELETE_TOPICS returns after the stamp. Brokers drain in the background.

Tests

AbstractControlPlaneTest covers both backends: fence after delete(commit, find-batches, get-log-info, listOffsets, deleteRecords, enforceRetention, getCrossTierLogStart, advanceCrossTierLogStartOffset, repairDisklessLog, pruneDisklessLogs, init), capped drain, empty-log cap, and producer_state cleanup after recreate (live readers hide leaked rows).

PurgeDeletedLogsJobTest: cap, empty-log drops, producer_state on both purge loops, concurrent jobs on a shared file, and two held-open purge_deleted_logs_v1 transactions interleaved so the second blocks on the file lock and is the one that marks deleting.

TopicPurgerMockedTest: idle sleep on a SKIP LOCKED miss, saturation when the cap is hit.

Test plan

  • :storage:inkless:test --tests AbstractControlPlaneTest --tests PurgeDeletedLogsJobTest --tests TopicPurgerMockedTest --tests DeleteTopicJobTest
  • Confirm DELETE_TOPICS returns while batches still exist, then TopicPurger drains them and FileCleaner removes the objects.

V28 vs earlier versions

Schema and one-off cleanup (new in V28):

ALTER TABLE logs
ADD COLUMN deleted_at TIMESTAMP WITH TIME ZONE;

CREATE INDEX logs_by_deleted_at_idx
    ON logs (deleted_at)
    WHERE deleted_at IS NOT NULL;

-- The old delete_topic_v1 dropped logs without touching producer_state. Those
-- orphans have no log row, so the purger never sees them.
DELETE FROM producer_state ps
WHERE NOT EXISTS (
    SELECT 1
    FROM logs l
    WHERE l.topic_id = ps.topic_id
        AND l.partition = ps.partition
);

purge_deleted_logs_v1 is new. The replaced functions below are
unified diffs against the last prior migration that defined each one.

CREATE FUNCTION purge_deleted_logs_v1(
    arg_now TIMESTAMP WITH TIME ZONE,
    arg_max_batches INT
)
RETURNS SETOF purge_deleted_logs_response_v1 LANGUAGE plpgsql VOLATILE AS $$
DECLARE
    l_batches_deleted BIGINT := 0;
    l_logs_purged INT := 0;
    l_files_marked INT := 0;
    l_more_remain BOOLEAN := FALSE;
    l_log RECORD;
    l_remaining BIGINT;
    l_deleted_this_log BIGINT;
    l_file_id BIGINT;
BEGIN
    DROP TABLE IF EXISTS purge_affected_files;
    CREATE TEMPORARY TABLE purge_affected_files (
        file_id BIGINT PRIMARY KEY
    ) ON COMMIT DROP;

    -- Drop empty deleted logs without spending the batch budget. The same per-cycle
    -- cap still applies: a wide empty topic (many partitions, no batches) must not
    -- take an unbounded number of row locks in one transaction. Leftovers keep
    -- more_remain true for the next cycle. 0 / NULL means unbounded.
    FOR l_log IN
        SELECT topic_id, partition
        FROM logs
        WHERE deleted_at IS NOT NULL
            AND NOT EXISTS (
                SELECT 1
                FROM batches b
                WHERE b.topic_id = logs.topic_id
                    AND b.partition = logs.partition
            )
        ORDER BY topic_id, partition
        LIMIT (CASE WHEN arg_max_batches > 0 THEN arg_max_batches END)
        FOR UPDATE SKIP LOCKED
    LOOP
        DELETE FROM producer_state
        WHERE topic_id = l_log.topic_id
            AND partition = l_log.partition;
        DELETE FROM logs
        WHERE topic_id = l_log.topic_id
            AND partition = l_log.partition;
        l_logs_purged := l_logs_purged + 1;
    END LOOP;

    l_remaining := CASE WHEN arg_max_batches > 0 THEN arg_max_batches::bigint ELSE NULL END;

    FOR l_log IN
        SELECT topic_id, partition
        FROM logs
        WHERE deleted_at IS NOT NULL
            AND EXISTS (
                SELECT 1
                FROM batches b
                WHERE b.topic_id = logs.topic_id
                    AND b.partition = logs.partition
            )
        ORDER BY topic_id, partition
        FOR UPDATE SKIP LOCKED
    LOOP
        EXIT WHEN l_remaining IS NOT NULL AND l_remaining <= 0;

        l_deleted_this_log := 0;
        FOR l_file_id IN
            DELETE FROM batches
            WHERE batch_id IN (
                SELECT batch_id
                FROM batches
                WHERE topic_id = l_log.topic_id
                    AND partition = l_log.partition
                ORDER BY topic_id, partition, last_offset
                LIMIT l_remaining
            )
            RETURNING file_id
        LOOP
            INSERT INTO purge_affected_files (file_id)
            VALUES (l_file_id)
            ON CONFLICT DO NOTHING;
            l_deleted_this_log := l_deleted_this_log + 1;
        END LOOP;

        l_batches_deleted := l_batches_deleted + l_deleted_this_log;
        IF l_remaining IS NOT NULL THEN
            l_remaining := l_remaining - l_deleted_this_log;
        END IF;

        IF NOT EXISTS (
            SELECT 1
            FROM batches
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition
            LIMIT 1
        ) THEN
            DELETE FROM producer_state
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition;
            DELETE FROM logs
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition;
            l_logs_purged := l_logs_purged + 1;
        END IF;
    END LOOP;

    -- Lock affected files in id order before the emptiness check. Two brokers can
    -- drain different partitions of the same WAL file (SKIP LOCKED on logs). Without
    -- this lock, neither transaction sees the other's uncommitted deletes, so both
    -- skip mark_file_to_delete_v1 and the file stays uploaded with zero batches.
    PERFORM 1
    FROM files
    WHERE file_id IN (SELECT file_id FROM purge_affected_files)
    ORDER BY file_id
    FOR UPDATE;

    SELECT COUNT(*)
    INTO l_files_marked
    FROM (
        SELECT DISTINCT af.file_id
        FROM purge_affected_files AS af
        WHERE NOT EXISTS (
            SELECT 1
            FROM batches AS b
            WHERE b.file_id = af.file_id
        )
    ) empty_files;

    PERFORM mark_file_to_delete_v1(arg_now, file_id)
    FROM (
        SELECT DISTINCT af.file_id
        FROM purge_affected_files AS af
        WHERE NOT EXISTS (
            SELECT 1
            FROM batches AS b
            WHERE b.file_id = af.file_id
        )
    ) empty_files;

    SELECT EXISTS (
        SELECT 1
        FROM logs
        WHERE deleted_at IS NOT NULL
    )
    INTO l_more_remain;

    RETURN NEXT (l_batches_deleted, l_logs_purged, l_files_marked, l_more_remain)::purge_deleted_logs_response_v1;
END;
$$
;

-- commit_file_v1 (from V20__Logs_earliest_batch_timestamp.sql): ignore soft-deleted logs.

delete_topic_v1 (was V6__Improve_understandability_of_delete_topic_v1.sql)

--- V6__Improve_understandability_of_delete_topic_v1.sql::delete_topic_v1
+++ V28__Soft_delete_logs.sql::delete_topic_v1
@@ -4,43 +4,19 @@
 )
 RETURNS VOID LANGUAGE plpgsql VOLATILE AS $$
 BEGIN
-    -- Ensure no other transaction commits or does anything else to the affected partitions while this transaction is in progress.
+    -- Wait for in-flight commits that already hold the log row, then stamp.
+    -- Later commits select only deleted_at IS NULL, so they see nonexistent_log.
     PERFORM
     FROM logs
     WHERE topic_id = ANY(arg_topic_ids)
-    ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
+        AND deleted_at IS NULL
+    ORDER BY topic_id, partition
     FOR UPDATE;

-    DROP TABLE IF EXISTS affected_files;
-    -- Delete the affected batches and remember what files are affected.
-    -- We need to separate deleting batches and finding empty files because if they are in the same requests as CTE,
-    -- the query below will see the MVCC snapshot from before deleting batches.
-    CREATE TEMPORARY TABLE affected_files
-    ON COMMIT DROP
-    AS
-        WITH deleted_batches AS (
-            DELETE FROM batches
-            WHERE topic_id = ANY(arg_topic_ids)
-            RETURNING file_id
-        )
-        SELECT file_id
-        FROM deleted_batches;
-
-    DELETE FROM logs
-    WHERE topic_id = ANY(arg_topic_ids);
-
-    -- Out of the affected files, select those that are now empty (i.e. no batch refers to them)
-    -- and mark them for deletion.
-    PERFORM mark_file_to_delete_v1(arg_now, file_id)
-    FROM (
-        SELECT DISTINCT af.file_id
-        FROM affected_files AS af
-        WHERE NOT EXISTS (
-            SELECT 1
            FROM batches AS b
            WHERE b.file_id = af.file_id
        )
    );
+    UPDATE logs
+    SET deleted_at = arg_now
+    WHERE topic_id = ANY(arg_topic_ids)
+        AND deleted_at IS NULL;
 END;
 $$
 ;

commit_file_v1 (was V20__Logs_earliest_batch_timestamp.sql)

--- V20__Logs_earliest_batch_timestamp.sql::commit_file_v1
+++ V28__Soft_delete_logs.sql::commit_file_v1
@@ -32,6 +32,7 @@
         SELECT *
         FROM logs
         WHERE (topic_id, partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
+            AND deleted_at IS NULL
         ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;

@@ -194,7 +195,8 @@
         LIMIT 1
     )
     WHERE (l.topic_id, l.partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
-        AND l.earliest_batch_timestamp IS NULL;
+        AND l.earliest_batch_timestamp IS NULL
+        AND l.deleted_at IS NULL;

     IF NOT EXISTS (SELECT 1 FROM batches WHERE file_id = l_new_file_id LIMIT 1) THEN
         PERFORM mark_file_to_delete_v1(arg_now, l_new_file_id);

commit_file_v2 (was V20__Logs_earliest_batch_timestamp.sql)

--- V20__Logs_earliest_batch_timestamp.sql::commit_file_v2
+++ V28__Soft_delete_logs.sql::commit_file_v2
@@ -48,6 +48,7 @@
         SELECT *
         FROM logs
         WHERE (topic_id, partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
+            AND deleted_at IS NULL
         ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;

@@ -233,7 +234,8 @@
         LIMIT 1
     )
     WHERE (l.topic_id, l.partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
-        AND l.earliest_batch_timestamp IS NULL;
+        AND l.earliest_batch_timestamp IS NULL
+        AND l.deleted_at IS NULL;

     IF NOT EXISTS (SELECT 1 FROM batches WHERE file_id = l_new_file_id LIMIT 1) THEN
         PERFORM mark_file_to_delete_v1(arg_now, l_new_file_id);

find_batches_v2 (was V27__Find_batches_request_level_min_one_message.sql)

--- V27__Find_batches_request_level_min_one_message.sql::find_batches_v2
+++ V28__Soft_delete_logs.sql::find_batches_v2
@@ -29,7 +29,7 @@
             END AS error
         FROM unnest(arg_requests) WITH ORDINALITY
             AS r(topic_id, partition, starting_offset, max_partition_fetch_bytes, ordinality)
-        LEFT JOIN logs l ON r.topic_id = l.topic_id AND r.partition = l.partition
+        LEFT JOIN logs l ON r.topic_id = l.topic_id AND r.partition = l.partition AND l.deleted_at IS NULL
         ORDER BY r.ordinality
     LOOP
         IF l_request.error IS NOT NULL THEN

delete_records_v1 (was V20__Logs_earliest_batch_timestamp.sql)

--- V20__Logs_earliest_batch_timestamp.sql::delete_records_v1
+++ V28__Soft_delete_logs.sql::delete_records_v1
@@ -25,6 +25,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         FOR UPDATE
         INTO l_log;

enforce_retention_v2 (was V22__Retention_enforcement_bounded_scan.sql)

--- V22__Retention_enforcement_bounded_scan.sql::enforce_retention_v2
+++ V28__Soft_delete_logs.sql::enforce_retention_v2
@@ -22,6 +22,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log; -- NOTE: no FOR UPDATE

         IF NOT FOUND THEN
@@ -63,6 +64,7 @@
                     FROM logs
                     WHERE topic_id = l_request.topic_id
                         AND partition = l_request.partition
+                        AND deleted_at IS NULL
                 ),
                 -- Scan only the oldest (max_batches_per_request + 1) batches. The retention boundary
                 -- cannot be deeper than max_batches_per_request because the delete is capped there
@@ -156,6 +158,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log
         FOR UPDATE;

list_offsets_v1 (was V17__List_offsets_cross_tier_earliest.sql)

--- V17__List_offsets_cross_tier_earliest.sql::list_offsets_v1
+++ V28__Soft_delete_logs.sql::list_offsets_v1
@@ -18,6 +18,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log;

         IF NOT FOUND THEN

init_diskless_log_v1 (was V23__Init_diskless_log_authoritative_seal.sql)

--- V23__Init_diskless_log_authoritative_seal.sql::init_diskless_log_v1
+++ V28__Soft_delete_logs.sql::init_diskless_log_v1
@@ -36,6 +36,7 @@
             WHERE logs.high_watermark = 0
                 AND logs.diskless_start_offset = 0
                 AND logs.byte_size = 0
+                AND logs.deleted_at IS NULL
                 -- A zero-offset re-init would not advance the placeholder, so leaving it alone keeps
                 -- re-initing an existing empty partition a no-op.
                 AND EXCLUDED.high_watermark > 0;

repair_diskless_log_v1 (was V18__Repair_diskless_log.sql)

--- V18__Repair_diskless_log.sql::repair_diskless_log_v1
+++ V28__Soft_delete_logs.sql::repair_diskless_log_v1
@@ -15,7 +15,8 @@
         UPDATE logs
             SET diskless_start_offset = l_request.diskless_start_offset
         WHERE topic_id = l_request.topic_id
-            AND partition = l_request.partition;
+            AND partition = l_request.partition
+            AND deleted_at IS NULL;

         l_found := FOUND;

advance_cross_tier_log_start_v1 (was V16__Cross_tier_log_start.sql)

--- V16__Cross_tier_log_start.sql::advance_cross_tier_log_start_v1
+++ V28__Soft_delete_logs.sql::advance_cross_tier_log_start_v1
@@ -1,4 +1,4 @@
-CREATE FUNCTION advance_cross_tier_log_start_v1(
+CREATE OR REPLACE FUNCTION advance_cross_tier_log_start_v1(
     arg_requests advance_cross_tier_log_start_request_v1[]
 )
 RETURNS SETOF advance_cross_tier_log_start_response_v1 LANGUAGE plpgsql VOLATILE AS $$
@@ -15,6 +15,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log
         FOR UPDATE;

prune_batches_below_highest_tiered_offset_v1 (was V20__Logs_earliest_batch_timestamp.sql)

--- V20__Logs_earliest_batch_timestamp.sql::prune_batches_below_highest_tiered_offset_v1
+++ V28__Soft_delete_logs.sql::prune_batches_below_highest_tiered_offset_v1
@@ -19,6 +19,7 @@
             FROM unnest(arg_requests) AS r
             WHERE r.topic_id = l.topic_id AND r.partition = l.partition
         )
+          AND l.deleted_at IS NULL
         ORDER BY l.topic_id, l.partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;
         FOREACH l_request IN ARRAY arg_requests LOOP
@@ -26,6 +27,7 @@
             FROM logs
             WHERE topic_id = l_request.topic_id
               AND partition = l_request.partition
+              AND deleted_at IS NULL
             ORDER BY topic_id, partition
             FOR UPDATE
             INTO l_log;

@viktorsomogyi
viktorsomogyi requested a lite review from Copilot September 3, 2026 09:34
Copilot AI reviewed Sep 3, 2026
View reviewed changes
Copilot AI left a comment
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Changes recommended

The new TopicPurgerMetrics gauges currently truncate LongAdder counters to 32-bit via intValue, which can overflow and misreport metrics on long-running brokers.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

This PR changes Inkless diskless-topic deletion from an eager, controller-thread purge into a soft-delete (logs.deleted_at) plus background purging on brokers. This bounds DELETE_TOPICS latency for large topics while still reclaiming batches/files asynchronously via a new purge_deleted_logs_v1 path and a broker TopicPurger.

Changes:

  • Add soft-delete semantics for diskless logs and introduce ControlPlane#purgeDeletedLogs(int) to purge soft-deleted logs with a per-cycle cap.
  • Add broker-side TopicPurger (scheduled from ReplicaManager) plus metrics/config knobs to control purge cadence and throughput.
  • Update Postgres and in-memory control planes, docs, and tests to reflect “stamp then purge” behavior and producer-state cleanup.
File summaries 8000
File Description
storage/inkless/src/test/java/io/aiven/inkless/delete/TopicPurgerMockedTest.java Adds unit coverage for TopicPurger idle/saturation/error behaviors.
storage/inkless/src/test/java/io/aiven/inkless/delete/FileCleanerIntegrationTest.java Ensures purged logs progress to file cleanup in integration flow.
storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/TopicsAndPartitionsCreateJobTest.java Updates expected logs records for new deleted_at column.
storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/PurgeDeletedLogsJobTest.java Adds integration coverage for Postgres purge logic and concurrency.
storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/DeleteTopicJobTest.java Updates delete-topic tests for soft-delete + explicit purge.
storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/DeleteRecordsJobTest.java Updates expected logs records for new deleted_at column.
storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/DeleteFilesJobTest.java Ensures deletes are followed by purge so file delete tests remain valid.
storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/DBUtils.java Adds helpers for reading producer_state in tests.
storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/CommitFileJobTest.java Updates expected logs records for new deleted_at column.
storage/inkless/src/test/java/io/aiven/inkless/control_plane/AbstractControlPlaneTest.java Expands control-plane contract tests for soft-delete fencing + purge.
storage/inkless/src/test/java/io/aiven/inkless/config/InklessConfigTest.java Adds config validation tests for topic purger interval/max-batches.
storage/inkless/src/main/jooq/org/jooq/generated/UDTs.java jOOQ regen for schema v28 (adds purge response UDT).
storage/inkless/src/main/jooq/org/jooq/generated/udt/RepairDisklessLogResponseV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/RepairDisklessLogRequestV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/RepairDisklessLogResponseV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/RepairDisklessLogRequestV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/PurgeDeletedLogsResponseV1Record.java New jOOQ record for purge response UDT.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/PruneBatchesBelowHighestTieredOffsetResponseV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/PruneBatchesBelowHighestTieredOffsetRequestV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/ListOffsetsResponseV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/ListOffsetsRequestV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/InitDisklessLogResponseV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/InitDisklessLogRequestV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/InitDisklessLogProducerStateV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/FindBatchesResponseV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/FindBatchesRequestV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/EnforceRetentionResponseV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/EnforceRetentionRequestV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/DeleteRecordsResponseV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/DeleteRecordsRequestV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/CommitBatchResponseV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/CommitBatchRequestV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/BatchMetadataV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/BatchInfoV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/AdvanceCrossTierLogStartResponseV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/AdvanceCrossTierLogStartRequestV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/PurgeDeletedLogsResponseV1.java New jOOQ UDT for purge response.
storage/inkless/src/main/jooq/org/jooq/generated/udt/PruneBatchesBelowHighestTieredOffsetResponseV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/PruneBatchesBelowHighestTieredOffsetRequestV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/RepairDisklessLogResponseV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/RepairDisklessLogRequestV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/PurgeDeletedLogsResponseV1Path.java New jOOQ path type for purge response UDT.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/PruneBatchesBelowHighestTieredOffsetResponseV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/PruneBatchesBelowHighestTieredOffsetRequestV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/ListOffsetsResponseV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/ListOffsetsRequestV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/InitDisklessLogResponseV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/InitDisklessLogRequestV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/InitDisklessLogProducerStateV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/FindBatchesResponseV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/FindBatchesRequestV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/EnforceRetentionResponseV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/EnforceRetentionRequestV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/DeleteRecordsResponseV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/DeleteRecordsRequestV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/CommitBatchResponseV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/CommitBatchRequestV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/BatchMetadataV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/BatchInfoV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/AdvanceCrossTierLogStartResponseV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/AdvanceCrossTierLogStartRequestV1Path.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/ListOffsetsResponseV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/ListOffsetsRequestV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/InitDisklessLogResponseV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/InitDisklessLogRequestV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/InitDisklessLogProducerStateV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/FindBatchesResponseV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/FindBatchesRequestV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/EnforceRetentionResponseV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/EnforceRetentionRequestV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/DeleteRecordsResponseV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/DeleteRecordsRequestV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/CommitBatchResponseV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/CommitBatchRequestV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/BatchMetadataV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/BatchInfoV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/AdvanceCrossTierLogStartResponseV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/udt/AdvanceCrossTierLogStartRequestV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/RepairDisklessLogV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/RepairDisklessLogV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/PurgeDeletedLogsV1Record.java New jOOQ record for purge TVF.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/PruneBatchesBelowHighestTieredOffsetV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/ProducerStateRecord.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/LogsRecord.java Adds deleted_at field mapping and ctor parameter.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/ListOffsetsV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/InitDisklessLogV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/FindBatchesV2Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/FindBatchesV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/FilesRecord.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/EnforceRetentionV2Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/EnforceRetentionV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/DeleteRecordsV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/CommitFileV2Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/CommitFileV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/BatchesRecord.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/AdvanceCrossTierLogStartV1Record.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/PurgeDeletedLogsV1.java New jOOQ table-valued function binding for purge_deleted_logs_v1.
storage/inkless/src/main/jooq/org/jooq/generated/tables/PruneBatchesBelowHighestTieredOffsetV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/ProducerState.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/Logs.java Adds deleted_at field and deleted-at index metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/ListOffsetsV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/InitDisklessLogV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/FindBatchesV2.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/FindBatchesV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/Files.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/EnforceRetentionV2.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/EnforceRetentionV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/DeleteRecordsV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/CommitFileV2.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/CommitFileV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/Batches.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/tables/AdvanceCrossTierLogStartV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/Tables.java Adds generated accessors for new purge TVF + schema v28 regen.
storage/inkless/src/main/jooq/org/jooq/generated/routines/MarkFileToDeleteV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/routines/FlushCommitRunV2.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/routines/DeleteTopicV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/routines/DeleteFilesV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/routines/DeleteBatchV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/routines/BatchTimestamp.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/Routines.java Adds generated routines for new purge TVF + schema v28 regen.
storage/inkless/src/main/jooq/org/jooq/generated/Keys.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/Indexes.java Adds generated index for logs_by_deleted_at_idx.
storage/inkless/src/main/jooq/org/jooq/generated/enums/PruneBatchesBelowHighestTieredOffsetErrorV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/enums/ListOffsetsResponseErrorV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/enums/InitDisklessLogErrorV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/enums/FindBatchesResponseErrorV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/enums/FileStateT.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/enums/FileReasonT.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/enums/EnforceRetentionResponseErrorV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/enums/DeleteRecordsResponseErrorV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/enums/CommitBatchResponseErrorV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/enums/AdvanceCrossTierLogStartResponseErrorV1.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/Domains.java jOOQ regen for schema v28.
storage/inkless/src/main/jooq/org/jooq/generated/DefaultSchema.java Adds new purge TVF/UDT to schema listing + schema v28 regen.
storage/inkless/src/main/java/io/aiven/inkless/doc/MetricsDocs.java Emits TopicPurger metrics into generated docs output.
storage/inkless/src/main/java/io/aiven/inkless/delete/TopicPurgerMetrics.java Adds broker-side purge-cycle metrics.
storage/inkless/src/main/java/io/aiven/inkless/delete/TopicPurger.java Adds scheduled broker worker to call purge and record metrics/backoff.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/PurgeDeletedLogsResponse.java Introduces shared response type for purge work accounting.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/postgres/PurgeDeletedLogsJob.java Adds Postgres job wrapper around purge_deleted_logs_v1.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/postgres/PostgresControlPlaneMetrics.java Adds query metrics tracking for purge calls.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/postgres/PostgresControlPlane.java Implements purgeDeletedLogs in Postgres control plane.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/postgres/GetProducerStateJob.java Filters out soft-deleted logs in producer-state lookup.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/postgres/GetLogInfoJob.java Filters out soft-deleted logs in log-info lookup.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/postgres/GetCrossTierLogStartJob.java Filters out soft-deleted logs in cross-tier start lookup.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/InMemoryControlPlane.java Mirrors soft-delete + purge + live-reader filtering in in-memory backend.
storage/inkless/src/main/java/io/aiven/inkless/control_plane/ControlPlane.java Adds purgeDeletedLogs(int) API to control-plane contract.
storage/inkless/src/main/java/io/aiven/inkless/config/InklessConfig.java Adds topic purger interval and max-batches configs with docs/defaults.
docs/inkless/metrics.rst Documents new TopicPurger metrics and new Postgres query metrics.
docs/inkless/FEATURES.md Documents DELETE_TOPICS returning after soft-delete stamp and background drain.
docs/inkless/configs.rst Documents new topic purger configuration keys.
core/src/main/scala/kafka/server/ReplicaManager.scala Schedules TopicPurger and closes it on shutdown.
Review details
  • Files reviewed: 29/152 changed files
  • Comments generated: 1
  • Review effort level: Lite

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

Comment thread storage/inkless/src/main/java/io/aiven/inkless/delete/TopicPurgerMetrics.java Resolved
@viktorsomogyi
viktorsomogyi marked this pull request as ready for review September 3, 2026 10:57
@viktorsomogyi
viktorsomogyi force-pushed the svv/diskless-async-delete branch from 4055773 to 4b5ab7b Compare September 3, 2026 11:02
@viktorsomogyi
viktorsomogyi force-pushed the svv/diskless-async-delete branch from 4b5ab7b to 868c6cf Compare September 3, 2026 15:06
giuseppelillo
giuseppelillo requested changes Sep 4, 2026
View reviewed changes
@giuseppelillo giuseppelillo left a comment
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just a couple of small issues

Comment thread storage/inkless/src/main/resources/db/migration/V28__Soft_delete_logs.sql Resolved
Comment thread storage/inkless/src/main/resources/db/migration/V28__Soft_delete_logs.sql Resolved
Comment thread storage/inkless/src/main/java/io/aiven/inkless/delete/TopicPurger.java Resolved
viktorsomogyi and others added 3 commits September 4, 2026 16:36
@viktorsomogyi @cursoragent
feat(inkless:delete): soft-delete diskless topics and purge in the ba…
e98fdcc 
…ckground [KC-349]

The old delete_topic_v1 deleted every batch for the topic on the
controller event-handler thread. A large diskless topic made
DELETE_TOPICS unbounded.

DELETE_TOPICS now only stamps logs.deleted_at. Live readers treat a
stamped row as gone. Brokers run TopicPurger, which calls
purge_deleted_logs_v1 with a cap. FileCleaner still deletes the objects
after file.cleaner.retention.period.ms.

## Controller path
delete_topic_v1 takes the live log rows FOR UPDATE, then sets
deleted_at. It does not delete batches, drop logs, or mark files.
ReplicaManager schedules inkless-topic-purger at
topic.purger.interval.ms (default 60s).

## Purge
An idle cycle returns after EXISTS(deleted_at) on the partial index,
before the temp table. That keeps an idle cluster from walking logs.

Empty deleted logs: drop producer_state, then the log. The same
per-cycle cap applies, so a wide empty topic cannot take an unbounded
number of row locks. Leftovers keep more_remain true.

Logs with batches: SKIP LOCKED, then the same shape as
enforce_retention_v2. Probe the (cap+1)-th last_offset on the covering
index (skip the probe when the cap is unbounded), range-delete
last_offset below that boundary, and drop producer_state and the log
when no batches remain. The leftover probe is EXISTS (... LIMIT 1),
not COUNT(*). Both drains order by deleted_at, topic_id, partition
so the partial index can drive them and older deletes go first. Loop 2
uses the same per-cycle LIMIT as loop 1.

Files: lock affected files by file_id before the emptiness check. Two
brokers can drain different partitions of one WAL file. Without that
lock, neither transaction sees the other's uncommitted deletes, both
skip mark_file_to_delete_v1, and the object stays uploaded with zero
batches.

more_remain is a cluster-wide EXISTS(deleted_at). TopicPurger treats a
cycle as idle when it deleted, purged, and marked nothing, so a SKIP
LOCKED miss does not spin. A cycle is saturated when batchesDeleted
reaches the cap and more_remain is still true.

## producer_state
The old delete_topic_v1 never removed producer_state, and there is no
foreign key to logs. A purge regression would restore that leak. Purge
deletes those rows when it drops a log. V28 also deletes orphans that
have no log row, because the purger only walks soft-deleted logs.

## Live readers
SQL functions that join or lock logs gain `deleted_at IS NULL`.
find_batches_v2 keeps V27's request-level l_global_bytes = 0 grant.
GetLogInfo, GetProducerState, and GetCrossTierLogStart apply the same
filter in Java. InMemoryControlPlane mirrors this with liveLog.

## Config and metrics
topic.purger.interval.ms defaults to 60000.
topic.purger.max.batches.per.cycle defaults to 20000. The same bound
caps empty-log drops. 0 means unbounded. With those defaults each
broker drains at most 20000 batches per minute.

TopicPurger metrics: cycle time and rate, batches/logs/files, errors,
TopicPurgerCycleSaturatedRate, and TopicPurgerWorkRemain (1 when the
last successful cycle left work).

FEATURES.md: DELETE_TOPICS returns after the stamp. Brokers drain in
the background.

## Tests
AbstractControlPlaneTest covers both backends: fence after delete
(commit, find-batches, get-log-info, listOffsets, deleteRecords,
enforceRetention, getCrossTierLogStart, advanceCrossTierLogStartOffset,
repairDisklessLog, pruneDisklessLogs, init), capped drain, FIFO by deleted_at, empty-log
cap, and producer_state cleanup after recreate (live readers hide
leaked rows).

PurgeDeletedLogsJobTest: cap, empty-log drops, producer_state on both
purge loops, concurrent jobs on a shared file, and two held-open
purge_deleted_logs_v1 transactions interleaved so the second blocks on
the file lock and is the one that marks deleting.

TopicPurgerMockedTest: idle sleep on a SKIP LOCKED miss, saturation
when the cap is hit.

## Test plan
- :storage:inkless:test --tests AbstractControlPlaneTest --tests PurgeDeletedLogsJobTest --tests TopicPurgerMockedTest --tests DeleteTopicJobTest
- Confirm DELETE_TOPICS returns while batches still exist, then TopicPurger drains them and FileCleaner removes the objects.

## V28 vs earlier versions

Schema and one-off cleanup (new in V28):

```sql
ALTER TABLE logs
ADD COLUMN deleted_at TIMESTAMP WITH TIME ZONE;

CREATE INDEX logs_by_deleted_at_idx
    ON logs (deleted_at)
    WHERE deleted_at IS NOT NULL;

-- The old delete_topic_v1 dropped logs without touching producer_state. Those
-- orphans have no log row, so the purger never sees them.
DELETE FROM producer_state ps
WHERE NOT EXISTS (
    SELECT 1
    FROM logs l
    WHERE l.topic_id = ps.topic_id
        AND l.partition = ps.partition
);
```

`purge_deleted_logs_v1` is new. The replaced functions below are
unified diffs against the last prior migration that defined each one.

```sql
CREATE FUNCTION purge_deleted_logs_v1(
    arg_now TIMESTAMP WITH TIME ZONE,
    arg_max_batches INT
)
RETURNS SETOF purge_deleted_logs_response_v1 LANGUAGE plpgsql VOLATILE AS $$
DECLARE
    l_batches_deleted BIGINT := 0;
    l_logs_purged INT := 0;
    l_files_marked INT := 0;
    l_more_remain BOOLEAN := FALSE;
    l_log RECORD;
    l_remaining BIGINT;
    l_deleted_this_log BIGINT;
    l_boundary offset_nullable_t;
BEGIN
    -- The partial index answers this probe. Without it, an idle cluster can walk
    -- logs_pkey filtering deleted_at on every live row, which is the produce hot path.
    IF NOT EXISTS (
        SELECT 1
        FROM logs
        WHERE deleted_at IS NOT NULL
    ) THEN
        RETURN NEXT (0, 0, 0, FALSE)::purge_deleted_logs_response_v1;
        RETURN;
    END IF;

    DROP TABLE IF EXISTS purge_affected_files;
    CREATE TEMPORARY TABLE purge_affected_files (
        file_id BIGINT PRIMARY KEY
    ) ON COMMIT DROP;

    -- Drop empty deleted logs without spending the batch budget. The same per-cycle
    -- cap still applies: a wide empty topic (many partitions, no batches) must not
    -- take an unbounded number of row locks in one transaction. Leftovers keep
    -- more_remain true for the next cycle. 0 / NULL means unbounded.
    -- deleted_at first so logs_by_deleted_at_idx drives the scan and older deletes go first.
    FOR l_log IN
        SELECT topic_id, partition
        FROM logs
        WHERE deleted_at IS NOT NULL
            AND NOT EXISTS (
                SELECT 1
                FROM batches b
                WHERE b.topic_id = logs.topic_id
                    AND b.partition = logs.partition
            )
        ORDER BY deleted_at, topic_id, partition
        LIMIT (CASE WHEN arg_max_batches > 0 THEN arg_max_batches END)
        FOR UPDATE SKIP LOCKED
    LOOP
        DELETE FROM producer_state
        WHERE topic_id = l_log.topic_id
            AND partition = l_log.partition;
        DELETE FROM logs
        WHERE topic_id = l_log.topic_id
            AND partition = l_log.partition;
        l_logs_purged := l_logs_purged + 1;
    END LOOP;

    l_remaining := CASE WHEN arg_max_batches > 0 THEN arg_max_batches::bigint ELSE NULL END;

    -- LIMIT is the batch budget: one batch per log is the worst case. Without it the planner
    -- can sort every soft-deleted log before the first DELETE.
    FOR l_log IN
        SELECT topic_id, partition
        FROM logs
        WHERE deleted_at IS NOT NULL
            AND EXISTS (
                SELECT 1
                FROM batches b
                WHERE b.topic_id = logs.topic_id
                    AND b.partition = logs.partition
            )
        ORDER BY deleted_at, topic_id, partition
        LIMIT (CASE WHEN arg_max_batches > 0 THEN arg_max_batches END)
        FOR UPDATE SKIP LOCKED
    LOOP
        EXIT WHEN l_remaining IS NOT NULL AND l_remaining <= 0;

        -- Same shape as enforce_retention_v2: probe the (cap+1)-th last_offset (index-only
        -- on the covering index), then range-delete. Selecting batch_id and deleting by
        -- PK heap-fetches every chosen row and descends batches_pkey again.
        l_boundary := NULL;
        IF l_remaining IS NOT NULL THEN
            SELECT last_offset
            FROM batches
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition
            ORDER BY topic_id, partition, last_offset
            LIMIT 1 OFFSET l_remaining
            INTO l_boundary;
        END IF;

        WITH deleted_batches AS (
            DELETE FROM batches
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition
                AND (l_boundary IS NULL OR last_offset < l_boundary)
            RETURNING file_id
        ),
        _1 AS (
            INSERT INTO purge_affected_files (file_id)
            SELECT DISTINCT file_id
            FROM deleted_batches
            ON CONFLICT DO NOTHING
        )
        SELECT COUNT(*)
        FROM deleted_batches
        INTO l_deleted_this_log;

        l_batches_deleted := l_batches_deleted + l_deleted_this_log;
        IF l_remaining IS NOT NULL THEN
            l_remaining := l_remaining - l_deleted_this_log;
        END IF;

        IF NOT EXISTS (
            SELECT 1
            FROM batches
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition
            LIMIT 1
        ) THEN
            DELETE FROM producer_state
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition;
            DELETE FROM logs
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition;
            l_logs_purged := l_logs_purged + 1;
        END IF;
    END LOOP;

    -- Lock affected files in id order before the emptiness check. Two brokers can
    -- drain different partitions of the same WAL file (SKIP LOCKED on logs). Without
    -- this lock, neither transaction sees the other's uncommitted deletes, so both
    -- skip mark_file_to_delete_v1 and the file stays uploaded with zero batches.
    PERFORM 1
    FROM files
    WHERE file_id IN (SELECT file_id FROM purge_affected_files)
    ORDER BY file_id
    FOR UPDATE;

    SELECT COUNT(*)
    INTO l_files_marked
    FROM (
        SELECT DISTINCT af.file_id
        FROM purge_affected_files AS af
        WHERE NOT EXISTS (
            SELECT 1
            FROM batches AS b
            WHERE b.file_id = af.file_id
        )
    ) empty_files;

    PERFORM mark_file_to_delete_v1(arg_now, file_id)
    FROM (
        SELECT DISTINCT af.file_id
        FROM purge_affected_files AS af
        WHERE NOT EXISTS (
            SELECT 1
            FROM batches AS b
            WHERE b.file_id = af.file_id
        )
    ) empty_files;

    SELECT EXISTS (
        SELECT 1
        FROM logs
        WHERE deleted_at IS NOT NULL
    )
    INTO l_more_remain;

    RETURN NEXT (l_batches_deleted, l_logs_purged, l_files_marked, l_more_remain)::purge_deleted_logs_response_v1;
END;
$$
;

-- commit_file_v1 (from V20__Logs_earliest_batch_timestamp.sql): ignore soft-deleted logs.
```

### `delete_topic_v1` (was V6__Improve_understandability_of_delete_topic_v1.sql)

```diff
--- V6__Improve_understandability_of_delete_topic_v1.sql::delete_topic_v1
+++ V28__Soft_delete_logs.sql::delete_topic_v1
@@ -4,43 +4,19 @@
 )
 RETURNS VOID LANGUAGE plpgsql VOLATILE AS $$
 BEGIN
-    -- Ensure no other transaction commits or does anything else to the affected partitions while this transaction is in progress.
+    -- Wait for in-flight commits that already hold the log row, then stamp.
+    -- Later commits select only deleted_at IS NULL, so they see nonexistent_log.
     PERFORM
     FROM logs
     WHERE topic_id = ANY(arg_topic_ids)
-    ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
+        AND deleted_at IS NULL
+    ORDER BY topic_id, partition
     FOR UPDATE;

-    DROP TABLE IF EXISTS affected_files;
-    -- Delete the affected batches and remember what files are affected.
-    -- We need to separate deleting batches and finding empty files because if they are in the same requests as CTE,
-    -- the query below will see the MVCC snapshot from before deleting batches.
-    CREATE TEMPORARY TABLE affected_files
-    ON COMMIT DROP
-    AS
-        WITH deleted_batches AS (
-            DELETE FROM batches
-            WHERE topic_id = ANY(arg_topic_ids)
-            RETURNING file_id
-        )
-        SELECT file_id
-        FROM deleted_batches;
-
-    DELETE FROM logs
-    WHERE topic_id = ANY(arg_topic_ids);
-
-    -- Out of the affected files, select those that are now empty (i.e. no batch refers to them)
-    -- and mark them for deletion.
-    PERFORM mark_file_to_delete_v1(arg_now, file_id)
-    FROM (
-        SELECT DISTINCT af.file_id
-        FROM affected_files AS af
-        WHERE NOT EXISTS (
-            SELECT 1
-            FROM batches AS b
-            WHERE b.file_id = af.file_id
-        )
-    );
+    UPDATE logs
+    SET deleted_at = arg_now
+    WHERE topic_id = ANY(arg_topic_ids)
+        AND deleted_at IS NULL;
 END;
 $$
 ;
```

### `commit_file_v1` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::commit_file_v1
+++ V28__Soft_delete_logs.sql::commit_file_v1
@@ -32,6 +32,7 @@
         SELECT *
         FROM logs
         WHERE (topic_id, partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
+            AND deleted_at IS NULL
         ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;

@@ -194,7 +195,8 @@
         LIMIT 1
     )
     WHERE (l.topic_id, l.partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
-        AND l.earliest_batch_timestamp IS NULL;
+        AND l.earliest_batch_timestamp IS NULL
+        AND l.deleted_at IS NULL;

     IF NOT EXISTS (SELECT 1 FROM batches WHERE file_id = l_new_file_id LIMIT 1) THEN
         PERFORM mark_file_to_delete_v1(arg_now, l_new_file_id);
```

### `commit_file_v2` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::commit_file_v2
+++ V28__Soft_delete_logs.sql::commit_file_v2
@@ -48,6 +48,7 @@
         SELECT *
         FROM logs
         WHERE (topic_id, partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
+            AND deleted_at IS NULL
         ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;

@@ -233,7 +234,8 @@
         LIMIT 1
     )
     WHERE (l.topic_id, l.partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
-        AND l.earliest_batch_timestamp IS NULL;
+        AND l.earliest_batch_timestamp IS NULL
+        AND l.deleted_at IS NULL;

     IF NOT EXISTS (SELECT 1 FROM batches WHERE file_id = l_new_file_id LIMIT 1) THEN
         PERFORM mark_file_to_delete_v1(arg_now, l_new_file_id);
```

### `find_batches_v2` (was V27__Find_batches_request_level_min_one_message.sql)

```diff
--- V27__Find_batches_request_level_min_one_message.sql::find_batches_v2
+++ V28__Soft_delete_logs.sql::find_batches_v2
@@ -29,7 +29,7 @@
             END AS error
         FROM unnest(arg_requests) WITH ORDINALITY
             AS r(topic_id, partition, starting_offset, max_partition_fetch_bytes, ordinality)
-        LEFT JOIN logs l ON r.topic_id = l.topic_id AND r.partition = l.partition
+        LEFT JOIN logs l ON r.topic_id = l.topic_id AND r.partition = l.partition AND l.deleted_at IS NULL
         ORDER BY r.ordinality
     LOOP
         IF l_request.error IS NOT NULL THEN
```

### `delete_records_v1` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::delete_records_v1
+++ V28__Soft_delete_logs.sql::delete_records_v1
@@ -25,6 +25,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         FOR UPDATE
         INTO l_log;

```

### `enforce_retention_v2` (was V22__Retention_enforcement_bounded_scan.sql)

```diff
--- V22__Retention_enforcement_bounded_scan.sql::enforce_retention_v2
+++ V28__Soft_delete_logs.sql::enforce_retention_v2
@@ -22,6 +22,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log; -- NOTE: no FOR UPDATE

         IF NOT FOUND THEN
@@ -63,6 +64,7 @@
                     FROM logs
                     WHERE topic_id = l_request.topic_id
                         AND partition = l_request.partition
+                        AND deleted_at IS NULL
                 ),
                 -- Scan only the oldest (max_batches_per_request + 1) batches. The retention boundary
                 -- cannot be deeper than max_batches_per_request because the delete is capped there
@@ -156,6 +158,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log
         FOR UPDATE;

```

### `list_offsets_v1` (was V17__List_offsets_cross_tier_earliest.sql)

```diff
--- V17__List_offsets_cross_tier_earliest.sql::list_offsets_v1
+++ V28__Soft_delete_logs.sql::list_offsets_v1
@@ -18,6 +18,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log;

         IF NOT FOUND THEN
```

### `init_diskless_log_v1` (was V23__Init_diskless_log_authoritative_seal.sql)

```diff
--- V23__Init_diskless_log_authoritative_seal.sql::init_diskless_log_v1
+++ V28__Soft_delete_logs.sql::init_diskless_log_v1
@@ -36,6 +36,7 @@
             WHERE logs.high_watermark = 0
                 AND logs.diskless_start_offset = 0
                 AND logs.byte_size = 0
+                AND logs.deleted_at IS NULL
                 -- A zero-offset re-init would not advance the placeholder, so leaving it alone keeps
                 -- re-initing an existing empty partition a no-op.
                 AND EXCLUDED.high_watermark > 0;
```

### `repair_diskless_log_v1` (was V18__Repair_diskless_log.sql)

```diff
--- V18__Repair_diskless_log.sql::repair_diskless_log_v1
+++ V28__Soft_delete_logs.sql::repair_diskless_log_v1
@@ -15,7 +15,8 @@
         UPDATE logs
             SET diskless_start_offset = l_request.diskless_start_offset
         WHERE topic_id = l_request.topic_id
-            AND partition = l_request.partition;
+            AND partition = l_request.partition
+            AND deleted_at IS NULL;

         l_found := FOUND;

```

### `advance_cross_tier_log_start_v1` (was V16__Cross_tier_log_start.sql)

```diff
--- V16__Cross_tier_log_start.sql::advance_cross_tier_log_start_v1
+++ V28__Soft_delete_logs.sql::advance_cross_tier_log_start_v1
@@ -1,4 +1,4 @@
-CREATE FUNCTION advance_cross_tier_log_start_v1(
+CREATE OR REPLACE FUNCTION advance_cross_tier_log_start_v1(
     arg_requests advance_cross_tier_log_start_request_v1[]
 )
 RETURNS SETOF advance_cross_tier_log_start_response_v1 LANGUAGE plpgsql VOLATILE AS $$
@@ -15,6 +15,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log
         FOR UPDATE;

```

### `prune_batches_below_highest_tiered_offset_v1` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::prune_batches_below_highest_tiered_offset_v1
+++ V28__Soft_delete_logs.sql::prune_batches_below_highest_tiered_offset_v1
@@ -19,6 +19,7 @@
             FROM unnest(arg_requests) AS r
             WHERE r.topic_id = l.topic_id AND r.partition = l.partition
         )
+          AND l.deleted_at IS NULL
         ORDER BY l.topic_id, l.partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;
         FOREACH l_request IN ARRAY arg_requests LOOP
@@ -26,6 +27,7 @@
             FROM logs
             WHERE topic_id = l_request.topic_id
               AND partition = l_request.partition
+              AND deleted_at IS NULL
             ORDER BY topic_id, partition
             FOR UPDATE
             INTO l_log;
```

Co-authored-by: Cursor <cursoragent@cursor.com>
@viktorsomogyi @cursoragent
fix(inkless:delete): lock a bounded purge window before classifying l…
10000
6b09ce5 
…ogs [KC-349]

EXISTS/NOT EXISTS before LIMIT walked every soft-deleted log to find
qualifying rows. The purger now locks the oldest N deleted logs, then
classifies them. logs_by_deleted_at_idx includes topic_id and partition
so that window is an index scan.

Co-authored-by: Cursor <cursoragent@cursor.com>
@viktorsomogyi @cursoragent
fix(inkless:delete): report cap_reached from purge_deleted_logs_v1 [K…
a6b172f 
…C-349]

TopicPurger treated a cycle as saturated when batchesDeleted reached the
cap. That missed a window of empty logs. The function now returns
cap_reached when more_remain is true and the call hit the window LIMIT
or the batch budget. Unbounded (0) stays false.

Co-authored-by: Cursor <cursoragent@cursor.com>
@viktorsomogyi
viktorsomogyi force-pushed the svv/diskless-async-delete branch from 868c6cf to a6b172f Compare September 4, 2026 14:37
@viktorsomogyi
viktorsomogyi commented Sep 4, 2026
Copy link
Copy Markdown
Contributor Author

@giuseppelillo thanks for the review pass, I added your requested changes.

@giuseppelillo
giuseppelillo merged commit a6716f7 into main Sep 4, 2026
5 checks passed
@giuseppelillo
giuseppelillo deleted the svv/diskless-async-delete branch September 4, 2026 15:54
jeqo pushed a commit that referenced this pull request Sep 9, 2026
@viktorsomogyi @cursoragent @jeqo
feat(inkless:delete): soft-delete diskless topics and purge in the ba…
ef7c6d4 
…ckground [KC-349] (#782)

* feat(inkless:delete): soft-delete diskless topics and purge in the background [KC-349]

The old delete_topic_v1 deleted every batch for the topic on the
controller event-handler thread. A large diskless topic made
DELETE_TOPICS unbounded.

DELETE_TOPICS now only stamps logs.deleted_at. Live readers treat a
stamped row as gone. Brokers run TopicPurger, which calls
purge_deleted_logs_v1 with a cap. FileCleaner still deletes the objects
after file.cleaner.retention.period.ms.

## Controller path
delete_topic_v1 takes the live log rows FOR UPDATE, then sets
deleted_at. It does not delete batches, drop logs, or mark files.
ReplicaManager schedules inkless-topic-purger at
topic.purger.interval.ms (default 60s).

## Purge
An idle cycle returns after EXISTS(deleted_at) on the partial index,
before the temp table. That keeps an idle cluster from walking logs.

Empty deleted logs: drop producer_state, then the log. The same
per-cycle cap applies, so a wide empty topic cannot take an unbounded
number of row locks. Leftovers keep more_remain true.

Logs with batches: SKIP LOCKED, then the same shape as
enforce_retention_v2. Probe the (cap+1)-th last_offset on the covering
index (skip the probe when the cap is unbounded), range-delete
last_offset below that boundary, and drop producer_state and the log
when no batches remain. The leftover probe is EXISTS (... LIMIT 1),
not COUNT(*). Both drains order by deleted_at, topic_id, partition
so the partial index can drive them and older deletes go first. Loop 2
uses the same per-cycle LIMIT as loop 1.

Files: lock affected files by file_id before the emptiness check. Two
brokers can drain different partitions of one WAL file. Without that
lock, neither transaction sees the other's uncommitted deletes, both
skip mark_file_to_delete_v1, and the object stays uploaded with zero
batches.

more_remain is a cluster-wide EXISTS(deleted_at). TopicPurger treats a
cycle as idle when it deleted, purged, and marked nothing, so a SKIP
LOCKED miss does not spin. A cycle is saturated when batchesDeleted
reaches the cap and more_remain is still true.

## producer_state
The old delete_topic_v1 never removed producer_state, and there is no
foreign key to logs. A purge regression would restore that leak. Purge
deletes those rows when it drops a log. V28 also deletes orphans that
have no log row, because the purger only walks soft-deleted logs.

## Live readers
SQL functions that join or lock logs gain `deleted_at IS NULL`.
find_batches_v2 keeps V27's request-level l_global_bytes = 0 grant.
GetLogInfo, GetProducerState, and GetCrossTierLogStart apply the same
filter in Java. InMemoryControlPlane mirrors this with liveLog.

## Config and metrics
topic.purger.interval.ms defaults to 60000.
topic.purger.max.batches.per.cycle defaults to 20000. The same bound
caps empty-log drops. 0 means unbounded. With those defaults each
broker drains at most 20000 batches per minute.

TopicPurger metrics: cycle time and rate, batches/logs/files, errors,
TopicPurgerCycleSaturatedRate, and TopicPurgerWorkRemain (1 when the
last successful cycle left work).

FEATURES.md: DELETE_TOPICS returns after the stamp. Brokers drain in
the background.

## Tests
AbstractControlPlaneTest covers both backends: fence after delete
(commit, find-batches, get-log-info, listOffsets, deleteRecords,
enforceRetention, getCrossTierLogStart, advanceCrossTierLogStartOffset,
repairDisklessLog, pruneDisklessLogs, init), capped drain, FIFO by deleted_at, empty-log
cap, and producer_state cleanup after recreate (live readers hide
leaked rows).

PurgeDeletedLogsJobTest: cap, empty-log drops, producer_state on both
purge loops, concurrent jobs on a shared file, and two held-open
purge_deleted_logs_v1 transactions interleaved so the second blocks on
the file lock and is the one that marks deleting.

TopicPurgerMockedTest: idle sleep on a SKIP LOCKED miss, saturation
when the cap is hit.

## Test plan
- :storage:inkless:test --tests AbstractControlPlaneTest --tests PurgeDeletedLogsJobTest --tests TopicPurgerMockedTest --tests DeleteTopicJobTest
- Confirm DELETE_TOPICS returns while batches still exist, then TopicPurger drains them and FileCleaner removes the objects.

## V28 vs earlier versions

Schema and one-off cleanup (new in V28):

```sql
ALTER TABLE logs
ADD COLUMN deleted_at TIMESTAMP WITH TIME ZONE;

CREATE INDEX logs_by_deleted_at_idx
    ON logs (deleted_at)
    WHERE deleted_at IS NOT NULL;

-- The old delete_topic_v1 dropped logs without touching producer_state. Those
-- orphans have no log row, so the purger never sees them.
DELETE FROM producer_state ps
WHERE NOT EXISTS (
    SELECT 1
    FROM logs l
    WHERE l.topic_id = ps.topic_id
        AND l.partition = ps.partition
);
```

`purge_deleted_logs_v1` is new. The replaced functions below are
unified diffs against the last prior migration that defined each one.

```sql
CREATE FUNCTION purge_deleted_logs_v1(
    arg_now TIMESTAMP WITH TIME ZONE,
    arg_max_batches INT
)
RETURNS SETOF purge_deleted_logs_response_v1 LANGUAGE plpgsql VOLATILE AS $$
DECLARE
    l_batches_deleted BIGINT := 0;
    l_logs_purged INT := 0;
    l_files_marked INT := 0;
    l_more_remain BOOLEAN := FALSE;
    l_log RECORD;
    l_remaining BIGINT;
    l_deleted_this_log BIGINT;
    l_boundary offset_nullable_t;
BEGIN
    -- The partial index answers this probe. Without it, an idle cluster can walk
    -- logs_pkey filtering deleted_at on every live row, which is the produce hot path.
    IF NOT EXISTS (
        SELECT 1
        FROM logs
        WHERE deleted_at IS NOT NULL
    ) THEN
        RETURN NEXT (0, 0, 0, FALSE)::purge_deleted_logs_response_v1;
        RETURN;
    END IF;

    DROP TABLE IF EXISTS purge_affected_files;
    CREATE TEMPORARY TABLE purge_affected_files (
        file_id BIGINT PRIMARY KEY
    ) ON COMMIT DROP;

    -- Drop empty deleted logs without spending the batch budget. The same per-cycle
    -- cap still applies: a wide empty topic (many partitions, no batches) must not
    -- take an unbounded number of row locks in one transaction. Leftovers keep
    -- more_remain true for the next cycle. 0 / NULL means unbounded.
    -- deleted_at first so logs_by_deleted_at_idx drives the scan and older deletes go first
10BC8
.
    FOR l_log IN
        SELECT topic_id, partition
        FROM logs
        WHERE deleted_at IS NOT NULL
            AND NOT EXISTS (
                SELECT 1
                FROM batches b
                WHERE b.topic_id = logs.topic_id
                    AND b.partition = logs.partition
            )
        ORDER BY deleted_at, topic_id, partition
        LIMIT (CASE WHEN arg_max_batches > 0 THEN arg_max_batches END)
        FOR UPDATE SKIP LOCKED
    LOOP
        DELETE FROM producer_state
        WHERE topic_id = l_log.topic_id
            AND partition = l_log.partition;
        DELETE FROM logs
        WHERE topic_id = l_log.topic_id
            AND partition = l_log.partition;
        l_logs_purged := l_logs_purged + 1;
    END LOOP;

    l_remaining := CASE WHEN arg_max_batches > 0 THEN arg_max_batches::bigint ELSE NULL END;

    -- LIMIT is the batch budget: one batch per log is the worst case. Without it the planner
    -- can sort every soft-deleted log before the first DELETE.
    FOR l_log IN
        SELECT topic_id, partition
        FROM logs
        WHERE deleted_at IS NOT NULL
            AND EXISTS (
                SELECT 1
                FROM batches b
                WHERE b.topic_id = logs.topic_id
                    AND b.partition = logs.partition
            )
        ORDER BY deleted_at, topic_id, partition
        LIMIT (CASE WHEN arg_max_batches > 0 THEN arg_max_batches END)
        FOR UPDATE SKIP LOCKED
    LOOP
        EXIT WHEN l_remaining IS NOT NULL AND l_remaining <= 0;

        -- Same shape as enforce_retention_v2: probe the (cap+1)-th last_offset (index-only
        -- on the covering index), then range-delete. Selecting batch_id and deleting by
        -- PK heap-fetches every chosen row and descends batches_pkey again.
        l_boundary := NULL;
        IF l_remaining IS NOT NULL THEN
            SELECT last_offset
            FROM batches
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition
            ORDER BY topic_id, partition, last_offset
            LIMIT 1 OFFSET l_remaining
            INTO l_boundary;
        END IF;

        WITH deleted_batches AS (
            DELETE FROM batches
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition
                AND (l_boundary IS NULL OR last_offset < l_boundary)
            RETURNING file_id
        ),
        _1 AS (
            INSERT INTO purge_affected_files (file_id)
            SELECT DISTINCT file_id
            FROM deleted_batches
            ON CONFLICT DO NOTHING
        )
        SELECT COUNT(*)
        FROM deleted_batches
        INTO l_deleted_this_log;

        l_batches_deleted := l_batches_deleted + l_deleted_this_log;
        IF l_remaining IS NOT NULL THEN
            l_remaining := l_remaining - l_deleted_this_log;
        END IF;

        IF NOT EXISTS (
            SELECT 1
            FROM batches
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition
            LIMIT 1
        ) THEN
            DELETE FROM producer_state
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition;
            DELETE FROM logs
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition;
            l_logs_purged := l_logs_purged + 1;
        END IF;
    END LOOP;

    -- Lock affected files in id order before the emptiness check. Two brokers can
    -- drain different partitions of the same WAL file (SKIP LOCKED on logs). Without
    -- this lock, neither transaction sees the other's uncommitted deletes, so both
    -- skip mark_file_to_delete_v1 and the file stays uploaded with zero batches.
    PERFORM 1
    FROM files
    WHERE file_id IN (SELECT file_id FROM purge_affected_files)
    ORDER BY file_id
    FOR UPDATE;

    SELECT COUNT(*)
    INTO l_files_marked
    FROM (
        SELECT DISTINCT af.file_id
        FROM purge_affected_files AS af
        WHERE NOT EXISTS (
            SELECT 1
            FROM batches AS b
            WHERE b.file_id = af.file_id
        )
    ) empty_files;

    PERFORM mark_file_to_delete_v1(arg_now, file_id)
    FROM (
        SELECT DISTINCT af.file_id
        FROM purge_affected_files AS af
        WHERE NOT EXISTS (
            SELECT 1
            FROM batches AS b
            WHERE b.file_id = af.file_id
        )
    ) empty_files;

    SELECT EXISTS (
        SELECT 1
        FROM logs
        WHERE deleted_at IS NOT NULL
    )
    INTO l_more_remain;

    RETURN NEXT (l_batches_deleted, l_logs_purged, l_files_marked, l_more_remain)::purge_deleted_logs_response_v1;
END;
$$
;

-- commit_file_v1 (from V20__Logs_earliest_batch_timestamp.sql): ignore soft-deleted logs.
```

### `delete_topic_v1` (was V6__Improve_understandability_of_delete_topic_v1.sql)

```diff
--- V6__Improve_understandability_of_delete_topic_v1.sql::delete_topic_v1
+++ V28__Soft_delete_logs.sql::delete_topic_v1
@@ -4,43 +4,19 @@
 )
 RETURNS VOID LANGUAGE plpgsql VOLATILE AS $$
 BEGIN
-    -- Ensure no other transaction commits or does anything else to the affected partitions while this transaction is in progress.
+    -- Wait for in-flight commits that already hold the log row, then stamp.
+    -- Later commits select only deleted_at IS NULL, so they see nonexistent_log.
     PERFORM
     FROM logs
     WHERE topic_id = ANY(arg_topic_ids)
-    ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
+        AND deleted_at IS NULL
+    ORDER BY topic_id, partition
     FOR UPDATE;

-    DROP TABLE IF EXISTS affected_files;
-    -- Delete the affected batches and remember what files are affected.
-    -- We need to separate deleting batches and finding empty files because if they are in the same requests as CTE,
-    -- the query below will see the MVCC snapshot from before deleting batches.
-    CREATE TEMPORARY TABLE affected_files
-    ON COMMIT DROP
-    AS
-        WITH deleted_batches AS (
-            DELETE FROM batches
-            WHERE topic_id = ANY(arg_topic_ids)
-            RETURNING file_id
-        )
-        SELECT file_id
-        FROM deleted_batches;
-
-    DELETE FROM logs
-    WHERE topic_id = ANY(arg_topic_ids);
-
-    -- Out of the affected files, select those that are now empty (i.e. no batch refers to them)
-    -- and mark them for deletion.
-    PERFORM mark_file_to_delete_v1(arg_now, file_id)
-    FROM (
-        SELECT DISTINCT af.file_id
-        FROM affected_files AS af
-        WHERE NOT EXISTS (
-            SELECT 1
-            FROM batches AS b
-            WHERE b.file_id = af.file_id
-        )
-    );
+    UPDATE logs
+    SET deleted_at = arg_now
+    WHERE topic_id = ANY(arg_topic_ids)
+        AND deleted_at IS NULL;
 END;
 $$
 ;
```

### `commit_file_v1` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::commit_file_v1
+++ V28__Soft_delete_logs.sql::commit_file_v1
@@ -32,6 +32,7 @@
         SELECT *
         FROM logs
         WHERE (topic_id, partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
+            AND deleted_at IS NULL
         ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;

@@ -194,7 +195,8 @@
         LIMIT 1
     )
     WHERE (l.topic_id, l.partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
-        AND l.earliest_batch_timestamp IS NULL;
+        AND l.earliest_batch_timestamp IS NULL
+        AND l.deleted_at IS NULL;

     IF NOT EXISTS (SELECT 1 FROM batches WHERE file_id = l_new_file_id LIMIT 1) THEN
         PERFORM mark_file_to_delete_v1(arg_now, l_new_file_id);
```

### `commit_file_v2` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::commit_file_v2
+++ V28__Soft_delete_logs.sql::commit_file_v2
@@ -48,6 +48,7 @@
         SELECT *
         FROM logs
         WHERE (topic_id, partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
+            AND deleted_at IS NULL
         ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;

@@ -233,7 +234,8 @@
         LIMIT 1
     )
     WHERE (l.topic_id, l.partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
-        AND l.earliest_batch_timestamp IS NULL;
+        AND l.earliest_batch_timestamp IS NULL
+        AND l.deleted_at IS NULL;

     IF NOT EXISTS (SELECT 1 FROM batches WHERE file_id = l_new_file_id LIMIT 1) THEN
         PERFORM mark_file_to_delete_v1(arg_now, l_new_file_id);
```

### `find_batches_v2` (was V27__Find_batches_request_level_min_one_message.sql)

```diff
--- V27__Find_batches_request_level_min_one_message.sql::find_batches_v2
+++ V28__Soft_delete_logs.sql::find_batches_v2
@@ -29,7 +29,7 @@
             END AS error
         FROM unnest(arg_requests) WITH ORDINALITY
             AS r(topic_id, partition, starting_offset, max_partition_fetch_bytes, ordinality)
-        LEFT JOIN logs l ON r.topic_id = l.topic_id AND r.partition = l.partition
+        LEFT JOIN logs l ON r.topic_id = l.topic_id AND r.partition = l.partition AND l.deleted_at IS NULL
         ORDER BY r.ordinality
     LOOP
         IF l_request.error IS NOT NULL THEN
```

### `delete_records_v1` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::delete_records_v1
+++ V28__Soft_delete_logs.sql::delete_records_v1
@@ -25,6 +25,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         FOR UPDATE
         INTO l_log;

```

### `enforce_retention_v2` (was V22__Retention_enforcement_bounded_scan.sql)

```diff
--- V22__Retention_enforcement_bounded_scan.sql::enforce_retention_v2
+++ V28__Soft_delete_logs.sql::enforce_retention_v2
@@ -22,6 +22,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log; -- NOTE: no FOR UPDATE

         IF NOT FOUND THEN
@@ -63,6 +64,7 @@
                     FROM logs
                     WHERE topic_id = l_request.topic_id
                         AND partition = l_request.partition
+                        AND deleted_at IS NULL
                 ),
                 -- Scan only the oldest (max_batches_per_request + 1) batches. The retention boundary
                 -- cannot be deeper than max_batches_per_request because the delete is capped there
@@ -156,6 +158,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log
         FOR UPDATE;

```

### `list_offsets_v1` (was V17__List_offsets_cross_tier_earliest.sql)

```diff
--- V17__List_offsets_cross_tier_earliest.sql::list_offsets_v1
+++ V28__Soft_delete_logs.sql::list_offsets_v1
@@ -18,6 +18,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log;

         IF NOT FOUND THEN
```

### `init_diskless_log_v1` (was V23__Init_diskless_log_authoritative_seal.sql)

```diff
--- V23__Init_diskless_log_authoritative_seal.sql::init_diskless_log_v1
+++ V28__Soft_delete_logs.sql::init_diskless_log_v1
@@ -36,6 +36,7 @@
             WHERE logs.high_watermark = 0
                 AND logs.diskless_start_offset = 0
                 AND logs.byte_size = 0
+                AND logs.deleted_at IS NULL
                 -- A zero-offset re-init would not advance the placeholder, so leaving it alone keeps
                 -- re-initing an existing empty partition a no-op.
                 AND EXCLUDED.high_watermark > 0;
```

### `repair_diskless_log_v1` (was V18__Repair_diskless_log.sql)

```diff
--- V18__Repair_diskless_log.sql::repair_diskless_log_v1
+++ V28__Soft_delete_logs.sql::repair_diskless_log_v1
@@ -15,7 +15,8 @@
         UPDATE logs
             SET diskless_start_offset = l_request.diskless_start_offset
         WHERE topic_id = l_request.topic_id
-            AND partition = l_request.partition;
+            AND partition = l_request.partition
+            AND deleted_at IS NULL;

         l_found := FOUND;

```

### `advance_cross_tier_log_start_v1` (was V16__Cross_tier_log_start.sql)

```diff
--- V16__Cross_tier_log_start.sql::advance_cross_tier_log_start_v1
+++ V28__Soft_delete_logs.sql::advance_cross_tier_log_start_v1
@@ -1,4 +1,4 @@
-CREATE FUNCTION advance_cross_tier_log_start_v1(
+CREATE OR REPLACE FUNCTION advance_cross_tier_log_start_v1(
     arg_requests advance_cross_tier_log_start_request_v1[]
 )
 RETURNS SETOF advance_cross_tier_log_start_response_v1 LANGUAGE plpgsql VOLATILE AS $$
@@ -15,6 +15,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log
         FOR UPDATE;

```

### `prune_batches_below_highest_tiered_offset_v1` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::prune_batches_below_highest_tiered_offset_v1
+++ V28__Soft_delete_logs.sql::prune_batches_below_highest_tiered_offset_v1
@@ -19,6 +19,7 @@
             FROM unnest(arg_requests) AS r
             WHERE r.topic_id = l.topic_id AND r.partition = l.partition
         )
+          AND l.deleted_at IS NULL
         ORDER BY l.topic_id, l.partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;
         FOREACH l_request IN ARRAY arg_requests LOOP
@@ -26,6 +27,7 @@
             FROM logs
             WHERE topic_id = l_request.topic_id
               AND partition = l_request.partition
+              AND deleted_at IS NULL
             ORDER BY topic_id, partition
             FOR UPDATE
             INTO l_log;
```

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(inkless:delete): lock a bounded purge window before classifying logs [KC-349]

EXISTS/NOT EXISTS before LIMIT walked every soft-deleted log to find
qualifying rows. The purger now locks the oldest N deleted logs, then
classifies them. logs_by_deleted_at_idx includes topic_id and partition
so that window is an index scan.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(inkless:delete): report cap_reached from purge_deleted_logs_v1 [KC-349]

TopicPurger treated a cycle as saturated when batchesDeleted reached the
cap. That missed a window of empty logs. The function now returns
cap_reached when more_remain is true and the call hit the window LIMIT
or the batch budget. Unbounded (0) stays false.

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Cursor <cursoragent@cursor.com>
jeqo pushed a commit that referenced this pull request Sep 9, 2026
@viktorsomogyi @cursoragent @jeqo
feat(inkless:delete): soft-delete diskless topics and purge in the ba…
1acf026 
…ckground [KC-349] (#782)

* feat(inkless:delete): soft-delete diskless topics and purge in the background [KC-349]

The old delete_topic_v1 deleted every batch for the topic on the
controller event-handler thread. A large diskless topic made
DELETE_TOPICS unbounded.

DELETE_TOPICS now only stamps logs.deleted_at. Live readers treat a
stamped row as gone. Brokers run TopicPurger, which calls
purge_deleted_logs_v1 with a cap. FileCleaner still deletes the objects
after file.cleaner.retention.period.ms.

## Controller path
delete_topic_v1 takes the live log rows FOR UPDATE, then sets
deleted_at. It does not delete batches, drop logs, or mark files.
ReplicaManager schedules inkless-topic-purger at
topic.purger.interval.ms (default 60s).

## Purge
An idle cycle returns after EXISTS(deleted_at) on the partial index,
before the temp table. That keeps an idle cluster from walking logs.

Empty deleted logs: drop producer_state, then the log. The same
per-cycle cap applies, so a wide empty topic cannot take an unbounded
number of row locks. Leftovers keep more_remain true.

Logs with batches: SKIP LOCKED, then the same shape as
enforce_retention_v2. Probe the (cap+1)-th last_offset on the covering
index (skip the probe when the cap is unbounded), range-delete
last_offset below that boundary, and drop producer_state and the log
when no batches remain. The leftover probe is EXISTS (... LIMIT 1),
not COUNT(*). Both drains order by deleted_at, topic_id, partition
so the partial index can drive them and older deletes go first. Loop 2
uses the same per-cycle LIMIT as loop 1.

Files: lock affected files by file_id before the emptiness check. Two
brokers can drain different partitions of one WAL file. Without that
lock, neither transaction sees the other's uncommitted deletes, both
skip mark_file_to_delete_v1, and the object stays uploaded with zero
batches.

more_remain is a cluster-wide EXISTS(deleted_at). TopicPurger treats a
cycle as idle when it deleted, purged, and marked nothing, so a SKIP
LOCKED miss does not spin. A cycle is saturated when batchesDeleted
reaches the cap and more_remain is still true.

## producer_state
The old delete_topic_v1 never removed producer_state, and there is no
foreign key to logs. A purge regression would restore that leak. Purge
deletes those rows when it drops a log. V28 also deletes orphans that
have no log row, because the purger only walks soft-deleted logs.

## Live readers
SQL functions that join or lock logs gain `deleted_at IS NULL`.
find_batches_v2 keeps V27's request-level l_global_bytes = 0 grant.
GetLogInfo, GetProducerState, and GetCrossTierLogStart apply the same
filter in Java. InMemoryControlPlane mirrors this with liveLog.

## Config and metrics
topic.purger.interval.ms defaults to 60000.
topic.purger.max.batches.per.cycle defaults to 20000. The same bound
caps empty-log drops. 0 means unbounded. With those defaults each
broker drains at most 20000 batches per minute.

TopicPurger metrics: cycle time and rate, batches/logs/files, errors,
TopicPurgerCycleSaturatedRate, and TopicPurgerWorkRemain (1 when the
last successful cycle left work).

FEATURES.md: DELETE_TOPICS returns after the stamp. Brokers drain in
the background.

## Tests
AbstractControlPlaneTest covers both backends: fence after delete
(commit, find-batches, get-log-info, listOffsets, deleteRecords,
enforceRetention, getCrossTierLogStart, advanceCrossTierLogStartOffset,
repairDisklessLog, pruneDisklessLogs, init), capped drain, FIFO by deleted_at, empty-log
cap, and producer_state cleanup after recreate (live readers hide
leaked rows).

PurgeDeletedLogsJobTest: cap, empty-log drops, producer_state on both
purge loops, concurrent jobs on a shared file, and two held-open
purge_deleted_logs_v1 transactions interleaved so the second blocks on
the file lock and is the one that marks deleting.

TopicPurgerMockedTest: idle sleep on a SKIP LOCKED miss, saturation
when the cap is hit.

## Test plan
- :storage:inkless:test --tests AbstractControlPlaneTest --tests PurgeDeletedLogsJobTest --tests TopicPurgerMockedTest --tests DeleteTopicJobTest
- Confirm DELETE_TOPICS returns while batches still exist, then TopicPurger drains them and FileCleaner removes the objects.

## V28 vs earlier versions

Schema and one-off cleanup (new in V28):

```sql
ALTER TABLE logs
ADD COLUMN deleted_at TIMESTAMP WITH TIME ZONE;

CREATE INDEX logs_by_deleted_at_idx
    ON logs (deleted_at)
    WHERE deleted_at IS NOT NULL;

-- The old delete_topic_v1 dropped logs without touching producer_state. Those
-- orphans have no log row, so the purger never sees them.
DELETE FROM producer_state ps
WHERE NOT EXISTS (
    SELECT 1
    FROM logs l
    WHERE l.topic_id = ps.topic_id
        AND l.partition = ps.partition
);
```

`purge_deleted_logs_v1` is new. The replaced functions below are
unified diffs against the last prior migration that defined each one.

```sql
CREATE FUNCTION purge_deleted_logs_v1(
    arg_now TIMESTAMP WITH TIME ZONE,
    arg_max_batches INT
)
RETURNS SETOF purge_deleted_logs_response_v1 LANGUAGE plpgsql VOLATILE AS $$
DECLARE
    l_batches_deleted BIGINT := 0;
    l_logs_purged INT := 0;
    l_files_marked INT := 0;
    l_more_remain BOOLEAN := FALSE;
    l_log RECORD;
    l_remaining BIGINT;
    l_deleted_this_log BIGINT;
    l_boundary offset_nullable_t;
BEGIN
    -- The partial index answers this probe. Without it, an idle cluster can walk
    -- logs_pkey filtering deleted_at on every live row, which is the produce hot path.
    IF NOT EXISTS (
        SELECT 1
        FROM logs
        WHERE deleted_at IS NOT NULL
    ) THEN
        RETURN NEXT (0, 0, 0, FALSE)::purge_deleted_logs_response_v1;
        RETURN;
    END IF;

    DROP TABLE IF EXISTS purge_affected_files;
    CREATE TEMPORARY TABLE purge_affected_files (
        file_id BIGINT PRIMARY KEY
    ) ON COMMIT DROP;

    -- Drop empty deleted logs without spending the batch budget. The same per-cycle
    -- cap still applies: a wide empty topic (many partitions, no batches) must not
    -- take an unbounded number of row locks in one transaction. Leftovers keep
    -- more_remain true for the next cycle. 0 / NULL means unbounded.
    -- deleted_at first so logs_by_deleted_at_idx drives the scan and older deletes go first.
    FOR l_log IN
        SELECT topic_id, partition
        FROM logs
        WHERE deleted_at IS NOT NULL
            AND NOT EXISTS (
                SELECT 1
                FROM batches b
                WHERE b.topic_id = logs.topic_id
                    AND b.partition = logs.partition
            )
        ORDER BY deleted_at, topic_id, partition
        LIMIT (CASE WHEN arg_max_batches > 0 THEN arg_max_batches END)
        FOR UPDATE SKIP LOCKED
    LOOP
        DELETE FROM producer_state
        WHERE topic_id = l_log.topic_id
            AND partition = l_log.partition;
        DELETE FROM logs
        WHERE topic_id = l_log.topic_id
            AND partition = l_log.partition;
        l_logs_purged := l_logs_purged + 1;
    END LOOP;

    l_remaining := CASE WHEN arg_max_batches > 0 THEN arg_max_batches::bigint ELSE NULL END;

    -- LIMIT is the batch budget: one batch per log is the worst case. Without it the planner
    -- can sort every soft-deleted log before the first DELETE.
    FOR l_log IN
        SELECT topic_id, partition
        FROM logs
        WHERE deleted_at IS NOT NULL
            AND EXISTS (
                SELECT 1
                FROM batches b
                WHERE b.topic_id = logs.topic_id
                    AND b.partition = logs.partition
            )
        ORDER BY deleted_at, topic_id, partition
        LIMIT (CASE WHEN arg_max_batches > 0 THEN arg_max_batches END)
        FOR UPDATE SKIP LOCKED
    LOOP
        EXIT WHEN l_remaining IS NOT NULL AND l_remaining <= 0;

        -- Same shape as enforce_retention_v2: probe the (cap+1)-th last_offset (index-only
        -- on the covering index), then range-delete. Selecting batch_id and deleting by
        -- PK heap-fetches every chosen row and descends batches_pkey again.
        l_boundary := NULL;
        IF l_remaining IS NOT NULL THEN
            SELECT last_offset
            FROM batches
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition
            ORDER BY topic_id, partition, last_offset
            LIMIT 1 OFFSET l_remaining
            INTO l_boundary;
        END IF;

        WITH deleted_batches AS (
            DELETE FROM batches
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition
                AND (l_boundary IS NULL OR last_offset < l_boundary)
            RETURNING file_id
        ),
        _1 AS (
            INSERT INTO purge_affected_files (file_id)
            SELECT DISTINCT file_id
            FROM deleted_batches
            ON CONFLICT DO NOTHING
        )
        SELECT COUNT(*)
        FROM deleted_batches
        INTO l_deleted_this_log;

        l_batches_deleted := l_batches_deleted + l_deleted_this_log;
        IF l_remaining IS NOT NULL THEN
            l_remaining := l_remaining - l_deleted_this_log;
        END IF;

        IF NOT EXISTS (
            SELECT 1
            FROM batches
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition
            LIMIT 1
        ) THEN
            DELETE FROM producer_state
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition;
            DELETE FROM logs
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition;
            l_logs_purged := l_logs_purged + 1;
        END IF;
    END LOOP;

    -- Lock affected files in id order before the emptiness check. Two brokers can
    -- drain different partitions of the same WAL file (SKIP LOCKED on logs). Without
    -- this lock, neither transaction sees the other's uncommitted deletes, so both
    -- skip mark_file_to_delete_v1 and the file stays uploaded with zero batches.
    PERFORM 1
    FROM files
    WHERE file_id IN (SELECT file_id FROM purge_affected_files)
    ORDER BY file_id
    FOR UPDATE;

    SELECT COUNT(*)
    INTO l_files_marked
    FROM (
        SELECT DISTINCT af.file_id
        FROM purge_affected_files AS af
        WHERE NOT EXISTS (
            SELECT 1
            FROM batches AS b
            WHERE b.file_id = af.file_id
        )
    ) empty_files;

    PERFORM mark_file_to_delete_v1(arg_now, file_id)
    FROM (
        SELECT DISTINCT af.file_id
        FROM purge_affected_files AS af
        WHERE NOT EXISTS (
            SELECT 1
            FROM batches AS b
            WHERE b.file_id = af.file_id
        )
    ) empty_files;

    SELECT EXISTS (
        SELECT 1
        FROM logs
        WHERE deleted_at IS NOT NULL
    )
    INTO l_more_remain;

    RETURN NEXT (l_batches_deleted, l_logs_purged, l_files_marked, l_more_remain)::purge_deleted_logs_response_v1;
END;
$$
;

-- commit_file_v1 (from V20__Logs_earliest_batch_timestamp.sql): ignore soft-deleted logs.
```

### `delete_topic_v1` (was V6__Improve_understandability_of_delete_topic_v1.sql)

```diff
--- V6__Improve_understandability_of_delete_topic_v1.sql::delete_topic_v1
+++ V28__Soft_delete_logs.sql::delete_topic_v1
@@ -4,43 +4,19 @@
 )
 RETURNS VOID LANGUAGE plpgsql VOLATILE AS $$
 BEGIN
-    -- Ensure no other transaction commits or does anything else to the affected partitions while this transaction is in progress.
+    -- Wait for in-flight commits that already hold the log row, then stamp.
+    -- Later commits select only deleted_at IS NULL, so they see nonexistent_log.
     PERFORM
     FROM logs
     WHERE topic_id = ANY(arg_topic_ids)
-    ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
+        AND deleted_at IS NULL
+    ORDER BY topic_id, partition
     FOR UPDATE;

-    DROP TABLE IF EXISTS affected_files;
-    -- Delete the affected batches and remember what files are affected.
-    -- We need to separate deleting batches and finding empty files because if they are in the same requests as CTE,
-    -- the query below will see the MVCC snapshot from before deleting batches.
-    CREATE TEMPORARY TABLE affected_files
-    ON COMMIT DROP
-    AS
-        WITH deleted_batches AS (
-            DELETE FROM batches
-            WHERE topic_id = ANY(arg_topic_ids)
-            RETURNING file_id
-        )
-        SELECT file_id
-        FROM deleted_batches;
-
-    DELETE FROM logs
-    WHERE topic_id = ANY(arg_topic_ids);
-
-    -- Out of the affected files, select those that are now empty (i.e. no batch refers to them)
-    -- and mark them for deletion.
-    PERFORM mark_file_to_delete_v1(arg_now, file_id)
-    FROM (
-        SELECT DISTINCT af.file_id
-        FROM affected_files AS af
-        WHERE NOT EXISTS (
-            SELECT 1
-            FROM batches AS b
-            WHERE b.file_id = af.file_id
-        )
-    );
+    UPDATE logs
+    SET deleted_at = arg_now
+    WHERE topic_id = ANY(arg_topic_ids)
+        AND deleted_at IS NULL;
 END;
 $$
 ;
```

### `commit_file_v1` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::commit_file_v1
+++ V28__Soft_delete_logs.sql::commit_file_v1
@@ -32,6 +32,7 @@
         SELECT *
         FROM logs
         WHERE (topic_id, partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
+            AND deleted_at IS NULL
         ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;

@@ -194,7 +195,8 @@
         LIMIT 1
     )
     WHERE (l.topic_id, l.partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
-        AND l.earliest_batch_timestamp IS NULL;
+        AND l.earliest_batch_timestamp IS NULL
+        AND l.deleted_at IS NULL;

     IF NOT EXISTS (SELECT 1 FROM batches WHERE file_id = l_new_file_id LIMIT 1) THEN
         PERFORM mark_file_to_delete_v1(arg_now, l_new_file_id);
```

### `commit_file_v2` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::commit_file_v2
+++ V28__Soft_delete_logs.sql::commit_file_v2
@@ -48,6 +48,7 @@
         SELECT *
         FROM logs
         WHERE (topic_id, partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
+            AND deleted_at IS NULL
         ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;

@@ -233,7 +234,8 @@
         LIMIT 1
     )
     WHERE (l.topic_id, l.partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
-        AND l.earliest_batch_timestamp IS NULL;
+        AND l.earliest_batch_timestamp IS NULL
+        AND l.deleted_at IS NULL;

     IF NOT EXISTS (SELECT 1 FROM batches WHERE file_id = l_new_file_id LIMIT 1) THEN
         PERFORM mark_file_to_delete_v1(arg_now, l_new_file_id);
```

### `find_batches_v2` (was V27__Find_batches_request_level_min_one_message.sql)

```diff
--- V27__Find_batches_request_level_min_one_message.sql::find_batches_v2
+++ V28__Soft_delete_logs.sql::find_batches_v2
@@ -29,7 +29,7 @@
             END AS error
         FROM unnest(arg_requests) WITH ORDINALITY
             AS r(topic_id, partition, starting_offset, max_partition_fetch_bytes, ordinality)
-        LEFT JOIN logs l ON r.topic_id = l.topic_id AND r.partition = l.partition
+        LEFT JOIN logs l ON r.topic_id = l.topic_id AND r.partition = l.partition AND l.deleted_at IS NULL
         ORDER BY r.ordinality
     LOOP
         IF l_request.error IS NOT NULL THEN
```

### `delete_records_v1` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::delete_records_v1
+++ V28__Soft_delete_logs.sql::delete_records_v1
@@ -25,6 +25,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         FOR UPDATE
         INTO l_log;

```

### `enforce_retention_v2` (was V22__Retention_enforcement_bounded_scan.sql)

```diff
--- V22__Retention_enforcement_bounded_scan.sql::enforce_retention_v2
+++ V28__Soft_delete_logs.sql::enforce_retention_v2
@@ -22,6 +22,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log; -- NOTE: no FOR UPDATE

         IF NOT FOUND THEN
@@ -63,6 +64,7 @@
                     FROM logs
                     WHERE topic_id = l_request.topic_id
                         AND partition = l_request.partition
+                        AND deleted_at IS NULL
                 ),
                 -- Scan only the oldest (max_batches_per_request + 1) batches. The retention boundary
                 -- cannot be deeper than max_batches_per_request because the delete is capped there
@@ -156,6 +158,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log
         FOR UPDATE;

```

### `list_offsets_v1` (was V17__List_offsets_cross_tier_earliest.sql)

```diff
--- V17__List_offsets_cross_tier_earliest.sql::list_offsets_v1
+++ V28__Soft_delete_logs.sql::list_offsets_v1
@@ -18,6 +18,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log;

         IF NOT FOUND THEN
```

### `init_diskless_log_v1` (was V23__Init_diskless_log_authoritative_seal.sql)

```diff
--- V23__Init_diskless_log_authoritative_seal.sql::init_diskless_log_v1
+++ V28__Soft_delete_logs.sql::init_diskless_log_v1
@@ -36,6 +36,7 @@
             WHERE logs.high_watermark = 0
                 AND logs.diskless_start_offset = 0
                 AND logs.byte_size = 0
+                AND logs.deleted_at IS NULL
                 -- A zero-offset re-init would not advance the placeholder, so leaving it alone keeps
                 -- re-initing an existing empty partition a no-op.
                 AND EXCLUDED.high_watermark > 0;
```

### `repair_diskless_log_v1` (was V18__Repair_diskless_log.sql)

```diff
--- V18__Repair_diskless_log.sql::repair_diskless_log_v1
+++ V28__Soft_delete_logs.sql::repair_diskless_log_v1
@@ -15,7 +15,8 @@
         UPDATE logs
             SET diskless_start_offset = l_request.diskless_start_offset
         WHERE topic_id = l_request.topic_id
-            AND partition = l_request.partition;
+            AND partition = l_request.partition
+            AND deleted_at IS NULL;

         l_found := FOUND;

```

### `advance_cross_tier_log_start_v1` (was V16__Cross_tier_log_start.sql)

```diff
--- V16__Cross_tier_log_start.sql::advance_cross_tier_log_start_v1
+++ V28__Soft_delete_logs.sql::advance_cross_tier_log_start_v1
@@ -1,4 +1,4 @@
-CREATE FUNCTION advance_cross_tier_log_start_v1(
+CREATE OR REPLACE FUNCTION advance_cross_tier_log_start_v1(
     arg_requests advance_cross_tier_log_start_request_v1[]
 )
 RETURNS SETOF advance_cross_tier_log_start_response_v1 LANGUAGE plpgsql VOLATILE AS $$
@@ -15,6 +15,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log
         FOR UPDATE;

```

### `prune_batches_below_highest_tiered_offset_v1` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::prune_batches_below_highest_tiered_offset_v1
+++ V28__Soft_delete_logs.sql::prune_batches_below_highest_tiered_offset_v1
@@ -19,6 +19,7 @@
             FROM unnest(arg_requests) AS r
             WHERE r.topic_id = l.topic_id AND r.partition = l.partition
         )
+          AND l.deleted_at IS NULL
         ORDER BY l.topic_id, l.partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;
         FOREACH l_request IN ARRAY arg_requests LOOP
@@ -26,6 +27,7 @@
             FROM logs
             WHERE topic_id = l_request.topic_id
               AND partition = l_request.partition
+              AND deleted_at IS NULL
             ORDER BY topic_id, partition
             FOR UPDATE
             INTO l_log;
```

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(inkless:delete): lock a bounded purge window before classifying logs [KC-349]

EXISTS/NOT EXISTS before LIMIT walked every soft-deleted log to find
qualifying rows. The purger now locks the oldest N deleted logs, then
classifies them. logs_by_deleted_at_idx includes topic_id and partition
so that window is an index scan.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(inkless:delete): report cap_reached from purge_deleted_logs_v1 [KC-349]

TopicPurger treated a cycle as saturated when batchesDeleted reached the
cap. That missed a window of empty logs. The function now returns
cap_reached when more_remain is true and the call hit the window LIMIT
or the batch budget. Unbounded (0) stays false.

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Cursor <cursoragent@cursor.com>
jeqo pushed a commit that referenced this pull request Sep 9, 2026
@viktorsomogyi @cursoragent @jeqo
feat(inkless:delete): soft-delete diskless topics and purge in the ba…
faa1d26 
…ckground [KC-349] (#782)

* feat(inkless:delete): soft-delete diskless topics and purge in the background [KC-349]

The old delete_topic_v1 deleted every batch for the topic on the
controller event-handler thread. A large diskless topic made
DELETE_TOPICS unbounded.

DELETE_TOPICS now only stamps logs.deleted_at. Live readers treat a
stamped row as gone. Brokers run TopicPurger, which calls
purge_deleted_logs_v1 with a cap. FileCleaner still deletes the objects
after file.cleaner.retention.period.ms.

## Controller path
delete_topic_v1 takes the live log rows FOR UPDATE, then sets
deleted_at. It does not delete batches, drop logs, or mark files.
ReplicaManager schedules inkless-topic-purger at
topic.purger.interval.ms (default 60s).

## Purge
An idle cycle returns after EXISTS(deleted_at) on the partial index,
before the temp table. That keeps an idle cluster from walking logs.

Empty deleted logs: drop producer_state, then the log. The same
per-cycle cap applies, so a wide empty topic cannot take an unbounded
number of row locks. Leftovers keep more_remain true.

Logs with batches: SKIP LOCKED, then the same shape as
enforce_retention_v2. Probe the (cap+1)-th last_offset on the covering
index (skip the probe when the cap is unbounded), range-delete
last_offset below that boundary, and drop producer_state and the log
when no batches remain. The leftover probe is EXISTS (... LIMIT 1),
not COUNT(*). Both drains order by deleted_at, topic_id, partition
so the partial index can drive them and older deletes go first. Loop 2
uses the same per-cycle LIMIT as loop 1.

Files: lock affected files by file_id before the emptiness check. Two
brokers can drain different partitions of one WAL file. Without that
lock, neither transaction sees the other's uncommitted deletes, both
skip mark_file_to_delete_v1, and the object stays uploaded with zero
batches.

more_remain is a cluster-wide EXISTS(deleted_at). TopicPurger treats a
cycle as idle when it deleted, purged, and marked nothing, so a SKIP
LOCKED miss does not spin. A cycle is saturated when batchesDeleted
reaches the cap and more_remain is still true.

## producer_state
The old delete_topic_v1 never removed producer_state, and there is no
foreign key to logs. A purge regression would restore that leak. Purge
deletes those rows when it drops a log. V28 also deletes orphans that
have no log row, because the purger only walks soft-deleted logs.

## Live readers
SQL functions that join or lock logs gain `deleted_at IS NULL`.
find_batches_v2 keeps V27's request-level l_global_bytes = 0 grant.
GetLogInfo, GetProducerState, and GetCrossTierLogStart apply the same
filter in Java. InMemoryControlPlane mirrors this with liveLog.

## Config and metrics
topic.purger.interval.ms defaults to 60000.
topic.purger.max.batches.per.cycle defaults to 20000. The same bound
caps empty-log drops. 0 means unbounded. With those defaults each
broker drains at most 20000 batches per minute.

TopicPurger metrics: cycle time and rate, batches/logs/files, errors,
TopicPurgerCycleSaturatedRate, and TopicPurgerWorkRemain (1 when the
last successful cycle left work).

FEATURES.md: DELETE_TOPICS returns after the stamp. Brokers drain in
the background.

## Tests
AbstractControlPlaneTest covers both backends: fence after delete
(commit, find-batches, get-log-info, listOffsets, deleteRecords,
enforceRetention, getCrossTierLogStart, advanceCrossTierLogStartOffset,
repairDisklessLog, pruneDisklessLogs, init), capped drain, FIFO by deleted_at, empty-log
cap, and producer_state cleanup after recreate (live readers hide
leaked rows).

PurgeDeletedLogsJobTest: cap, empty-log drops, producer_state on both
purge loops, concurrent jobs on a shared file, and two held-open
purge_deleted_logs_v1 transactions interleaved so the second blocks on
the file lock and is the one that marks deleting.

TopicPurgerMockedTest: idle sleep on a SKIP LOCKED miss, saturation
when the cap is hit.

## Test plan
- :storage:inkless:test --tests AbstractControlPlaneTest --tests PurgeDeletedLogsJobTest --tests TopicPurgerMockedTest --tests DeleteTopicJobTest
- Confirm DELETE_TOPICS returns while batches still exist, then TopicPurger drains them and FileCleaner removes the objects.

## V28 vs earlier versions

Schema and one-off cleanup (new in V28):

```sql
ALTER TABLE logs
ADD COLUMN deleted_at TIMESTAMP WITH TIME ZONE;

CREATE INDEX logs_by_deleted_at_idx
    ON logs (deleted_at)
    WHERE deleted_at IS NOT NULL;

-- The old delete_topic_v1 dropped logs without touching producer_state. Those
-- orphans have no log row, so the purger never sees them.
DELETE FROM producer_state ps
WHERE NOT EXISTS (
    SELECT 1
    FROM logs l
    WHERE l.topic_id = ps.topic_id
        AND l.partition = ps.partition
);
```

`purge_deleted_logs_v1` is new. The replaced functions below are
unified diffs against the last prior migration that defined each one.

```sql
CREATE FUNCTION purge_deleted_logs_v1(
    arg_now TIMESTAMP WITH TIME ZONE,
    arg_max_batches INT
)
RETURNS SETOF purge_deleted_logs_response_v1 LANGUAGE plpgsql VOLATILE AS $$
DECLARE
    l_batches_deleted BIGINT := 0;
    l_logs_purged INT := 0;
    l_files_marked INT := 0;
    l_more_remain BOOLEAN := FALSE;
    l_log RECORD;
    l_remaining BIGINT;
    l_deleted_this_log BIGINT;
    l_boundary offset_nullable_t;
BEGIN
    -- The partial index answers this probe. Without it, an idle cluster can walk
    -- logs_pkey filtering deleted_at on every live row, which is the produce hot path.
    IF NOT EXISTS (
        SELECT 1
        FROM logs
        WHERE deleted_at IS NOT NULL
    ) THEN
        RETURN NEXT (0, 0, 0, FALSE)::purge_deleted_logs_response_v1;
        RETURN;
    END IF;

    DROP TABLE IF EXISTS purge_affected_files;
    CREATE TEMPORARY TABLE purge_affected_files (
        file_id BIGINT PRIMARY KEY
    ) ON COMMIT DROP;

    -- Drop empty deleted logs without spending the batch budget. The same per-cycle
    -- cap still applies: a wide empty topic (many partitions, no batches) must not
    -- take an unbounded number of row locks in one transaction. Leftovers keep
    -- more_remain true for the next cycle. 0 / NULL means unbounded.
    -- deleted_at first so logs_by_deleted_at_idx drives the scan and older deletes go first.
    FOR l_log IN
        SELECT topic_id, partition
        FROM logs
        WHERE deleted_at IS NOT NULL
            AND NOT EXISTS (
                SELECT 1
                FROM batches b
                WHERE b.topic_id = logs.topic_id
                    AND b.partition = logs.partition
            )
        ORDER BY deleted_at, topic_id, partition
        LIMIT (CASE WHEN arg_max_batches > 0 THEN arg_max_batches END)
        FOR UPDATE SKIP LOCKED
    LOOP
        DELETE FROM producer_state
        WHERE topic_id = l_log.topic_id
            AND partition = l_log.partition;
        DELETE FROM logs
        WHERE topic_id = l_log.topic_id
            AND partition = l_log.partition;
        l_logs_purged := l_logs_purged + 1;
    END LOOP;

    l_remaining := CASE WHEN arg_max_batches > 0 THEN arg_max_batches::bigint ELSE NULL END;

    -- LIMIT is the batch budget: one batch per log is the worst case. Without it the planner
    -- can sort every soft-deleted log before the first DELETE.
    FOR l_log IN
        SELECT topic_id, partition
        FROM logs
        WHERE deleted_at IS NOT NULL
            AND EXISTS (
                SELECT 1
                FROM batches b
                WHERE b.topic_id = logs.topic_id
                    AND b.partition = logs.partition
            )
        ORDER BY deleted_at, topic_id, partition
        LIMIT (CASE WHEN arg_max_batches > 0 THEN arg_max_batches END)
        FOR UPDATE SKIP LOCKED
    LOOP
        EXIT WHEN l_remaining IS NOT NULL AND l_remaining <= 0;

        -- Same shape as enforce_retention_v2: probe the (cap+1)-th last_offset (index-only
        -- on the covering index), then range-delete. Selecting batch_id and deleting by
        -- PK heap-fetches every chosen row and descends batches_pkey again.
        l_boundary := NULL;
        IF l_remaining IS NOT NULL THEN
            SELECT last_offset
            FROM batches
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition
            ORDER BY topic_id, partition, last_offset
            LIMIT 1 OFFSET l_remaining
            INTO l_boundary;
        END IF;

        WITH deleted_batches AS (
            DELETE FROM batches
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition
                AND (l_boundary IS NULL OR last_offset < l_boundary)
            RETURNING file_id
        ),
        _1 AS (
            INSERT INTO purge_affected_files (file_id)
            SELECT DISTINCT file_id
            FROM deleted_batches
            ON CONFLICT DO NOTHING
        )
        SELECT COUNT(*)
        FROM deleted_batches
        INTO l_deleted_this_log;

        l_batches_deleted := l_batches_deleted + l_deleted_this_log;
        IF l_remaining IS NOT NULL THEN
            l_remaining := l_remaining - l_deleted_this_log;
        END IF;

        IF NOT EXISTS (
            SELECT 1
            FROM batches
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition
            LIMIT 1
        ) THEN
            DELETE FROM producer_state
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition;
            DELETE FROM logs
            WHERE topic_id = l_log.topic_id
                AND partition = l_log.partition;
            l_logs_purged := l_logs_purged + 1;
        END IF;
    END LOOP;

    -- Lock affected files in id order before the emptiness check. Two brokers can
    -- drain different partitions of the same WAL file (SKIP LOCKED on logs). Without
    -- this lock, neither transaction sees the other's uncommitted deletes, so both
    -- skip mark_file_to_delete_v1 and the file stays uploaded with zero batches.
    PERFORM 1
    FROM files
    WHERE file_id IN (SELECT file_id FROM purge_affected_files)
    ORDER BY file_id
    FOR UPDATE;

    SELECT COUNT(*)
    INTO l_files_marked
    FROM (
        SELECT DISTINCT af.file_id
        FROM purge_affected_files AS af
        WHERE NOT EXISTS (
            SELECT 1
            FROM batches AS b
            WHERE b.file_id = af.file_id
        )
    ) empty_files;

    PERFORM mark_file_to_delete_v1(arg_now, file_id)
    FROM (
        SELECT DISTINCT af.file_id
        FROM purge_affected_files AS af
        WHERE NOT EXISTS (
            SELECT 1
            FROM batches AS b
            WHERE b.file_id = af.file_id
        )
    ) empty_files;

    SELECT EXISTS (
        SELECT 1
        FROM logs
        WHERE deleted_at IS NOT NULL
    )
    INTO l_more_remain;

    RETURN NEXT (l_batches_deleted, l_logs_purged, l_files_marked, l_more_remain)::purge_deleted_logs_response_v1;
END;
$$
;

-- commit_file_v1 (from V20__Logs_earliest_batch_timestamp.sql): ignore soft-deleted logs.
```

### `delete_topic_v1` (was V6__Improve_understandability_of_delete_topic_v1.sql)

```diff
--- V6__Improve_understandability_of_delete_topic_v1.sql::delete_topic_v1
+++ V28__Soft_delete_logs.sql::delete_topic_v1
@@ -4,43 +4,19 @@
 )
 RETURNS VOID LANGUAGE plpgsql VOLATILE AS $$
 BEGIN
-    -- Ensure no other transaction commits or does anything else to the affected partitions while this transaction is in progress.
+    -- Wait for in-flight commits that already hold the log row, then stamp.
+    -- Later commits select only deleted_at IS NULL, so they see nonexistent_log.
     PERFORM
     FROM logs
     WHERE topic_id = ANY(arg_topic_ids)
-    ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
+        AND deleted_at IS NULL
+    ORDER BY topic_id, partition
     FOR UPDATE;

-    DROP TABLE IF EXISTS affected_files;
-    -- Delete the affected batches and remember what files are affected.
-    -- We need to separate deleting batches and finding empty files because if they are in the same requests as CTE,
-    -- the query below will see the MVCC snapshot from before deleting batches.
-    CREATE TEMPORARY TABLE affected_files
-    ON COMMIT DROP
-    AS
-        WITH deleted_batches AS (
-            DELETE FROM batches
-            WHERE topic_id = ANY(arg_topic_ids)
-            RETURNING file_id
-        )
-        SELECT file_id
-        FROM deleted_batches;
-
-    DELETE FROM logs
-    WHERE topic_id = ANY(arg_topic_ids);
-
-    -- Out of the affected files, select those that are now empty (i.e. no batch refers to them)
-    -- and mark them for deletion.
-    PERFORM mark_file_to_delete_v1(arg_now, file_id)
-    FROM (
-        SELECT DISTINCT af.file_id
-        FROM affected_files AS af
-        WHERE NOT EXISTS (
-            SELECT 1
-            FROM batches AS b
-            WHERE b.file_id = af.file_id
-        )
-    );
+    UPDATE logs
+    SET deleted_at = arg_now
+    WHERE topic_id = ANY(arg_topic_ids)
+        AND deleted_at IS NULL;
 END;
 $$
 ;
```

### `commit_file_v1` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::commit_file_v1
+++ V28__Soft_delete_logs.sql::commit_file_v1
@@ -32,6 +32,7 @@
         SELECT *
         FROM logs
         WHERE (topic_id, partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
+            AND deleted_at IS NULL
         ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;

@@ -194,7 +195,8 @@
         LIMIT 1
     )
     WHERE (l.topic_id, l.partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
-        AND l.earliest_batch_timestamp IS NULL;
+        AND l.earliest_batch_timestamp IS NULL
+        AND l.deleted_at IS NULL;

     IF NOT EXISTS (SELECT 1 FROM batches WHERE file_id = l_new_file_id LIMIT 1) THEN
         PERFORM mark_file_to_delete_v1(arg_now, l_new_file_id);
```

### `commit_file_v2` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::commit_file_v2
+++ V28__Soft_delete_logs.sql::commit_file_v2
@@ -48,6 +48,7 @@
         SELECT *
         FROM logs
         WHERE (topic_id, partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
+            AND deleted_at IS NULL
         ORDER BY topic_id, partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;

@@ -233,7 +234,8 @@
         LIMIT 1
     )
     WHERE (l.topic_id, l.partition) IN (SELECT DISTINCT topic_id, partition FROM unnest(arg_requests))
-        AND l.earliest_batch_timestamp IS NULL;
+        AND l.earliest_batch_timestamp IS NULL
+        AND l.deleted_at IS NULL;

     IF NOT EXISTS (SELECT 1 FROM batches WHERE file_id = l_new_file_id LIMIT 1) THEN
         PERFORM mark_file_to_delete_v1(arg_now, l_new_file_id);
```

### `find_batches_v2` (was V27__Find_batches_request_level_min_one_message.sql)

```diff
--- V27__Find_batches_request_level_min_one_message.sql::find_batches_v2
+++ V28__Soft_delete_logs.sql::find_batches_v2
@@ -29,7 +29,7 @@
             END AS error
         FROM unnest(arg_requests) WITH ORDINALITY
             AS r(topic_id, partition, starting_offset, max_partition_fetch_bytes, ordinality)
-        LEFT JOIN logs l ON r.topic_id = l.topic_id AND r.partition = l.partition
+        LEFT JOIN logs l ON r.topic_id = l.topic_id AND r.partition = l.partition AND l.deleted_at IS NULL
         ORDER BY r.ordinality
     LOOP
         IF l_request.error IS NOT NULL THEN
```

### `delete_records_v1` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::delete_records_v1
+++ V28__Soft_delete_logs.sql::delete_records_v1
@@ -25,6 +25,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         FOR UPDATE
         INTO l_log;

```

### `enforce_retention_v2` (was V22__Retention_enforcement_bounded_scan.sql)

```diff
--- V22__Retention_enforcement_bounded_scan.sql::enforce_retention_v2
+++ V28__Soft_delete_logs.sql::enforce_retention_v2
@@ -22,6 +22,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log; -- NOTE: no FOR UPDATE

         IF NOT FOUND THEN
@@ -63,6 +64,7 @@
                     FROM logs
                     WHERE topic_id = l_request.topic_id
                         AND partition = l_request.partition
+                        AND deleted_at IS NULL
                 ),
                 -- Scan only the oldest (max_batches_per_request + 1) batches. The retention boundary
                 -- cannot be deeper than max_batches_per_request because the delete is capped there
@@ -156,6 +158,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log
         FOR UPDATE;

```

### `list_offsets_v1` (was V17__List_offsets_cross_tier_earliest.sql)

```diff
--- V17__List_offsets_cross_tier_earliest.sql::list_offsets_v1
+++ V28__Soft_delete_logs.sql::list_offsets_v1
@@ -18,6 +18,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log;

         IF NOT FOUND THEN
```

### `init_diskless_log_v1` (was V23__Init_diskless_log_authoritative_seal.sql)

```diff
--- V23__Init_diskless_log_authoritative_seal.sql::init_diskless_log_v1
+++ V28__Soft_delete_logs.sql::init_diskless_log_v1
@@ -36,6 +36,7 @@
             WHERE logs.high_watermark = 0
                 AND logs.diskless_start_offset = 0
                 AND logs.byte_size = 0
+                AND logs.deleted_at IS NULL
                 -- A zero-offset re-init would not advance the placeholder, so leaving it alone keeps
                 -- re-initing an existing empty partition a no-op.
                 AND EXCLUDED.high_watermark > 0;
```

### `repair_diskless_log_v1` (was V18__Repair_diskless_log.sql)

```diff
--- V18__Repair_diskless_log.sql::repair_diskless_log_v1
+++ V28__Soft_delete_logs.sql::repair_diskless_log_v1
@@ -15,7 +15,8 @@
         UPDATE logs
             SET diskless_start_offset = l_request.diskless_start_offset
         WHERE topic_id = l_request.topic_id
-            AND partition = l_request.partition;
+            AND partition = l_request.partition
+            AND deleted_at IS NULL;

         l_found := FOUND;

```

### `advance_cross_tier_log_start_v1` (was V16__Cross_tier_log_start.sql)

```diff
--- V16__Cross_tier_log_start.sql::advance_cross_tier_log_start_v1
+++ V28__Soft_delete_logs.sql::advance_cross_tier_log_start_v1
@@ -1,4 +1,4 @@
-CREATE FUNCTION advance_cross_tier_log_start_v1(
+CREATE OR REPLACE FUNCTION advance_cross_tier_log_start_v1(
     arg_requests advance_cross_tier_log_start_request_v1[]
 )
 RETURNS SETOF advance_cross_tier_log_start_response_v1 LANGUAGE plpgsql VOLATILE AS $$
@@ -15,6 +15,7 @@
         FROM logs
         WHERE topic_id = l_request.topic_id
             AND partition = l_request.partition
+            AND deleted_at IS NULL
         INTO l_log
         FOR UPDATE;

```

### `prune_batches_below_highest_tiered_offset_v1` (was V20__Logs_earliest_batch_timestamp.sql)

```diff
--- V20__Logs_earliest_batch_timestamp.sql::prune_batches_below_highest_tiered_offset_v1
+++ V28__Soft_delete_logs.sql::prune_batches_below_highest_tiered_offset_v1
@@ -19,6 +19,7 @@
             FROM unnest(arg_requests) AS r
             WHERE r.topic_id = l.topic_id AND r.partition = l.partition
         )
+          AND l.deleted_at IS NULL
         ORDER BY l.topic_id, l.partition  -- ordering is important to prevent deadlocks
         FOR UPDATE;
         FOREACH l_request IN ARRAY arg_requests LOOP
@@ -26,6 +27,7 @@
             FROM logs
             WHERE topic_id = l_request.topic_id
               AND partition = l_request.partition
+              AND deleted_at IS NULL
             ORDER BY topic_id, partition
             FOR UPDATE
             INTO l_log;
```

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(inkless:delete): lock a bounded purge window before classifying logs [KC-349]

EXISTS/NOT EXISTS before LIMIT walked every soft-deleted log to find
qualifying rows. The purger now locks the oldest N deleted logs, then
classifies them. logs_by_deleted_at_idx includes topic_id and partition
so that window is an index scan.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(inkless:delete): report cap_reached from purge_deleted_logs_v1 [KC-349]

TopicPurger treated a cycle as saturated when batchesDeleted reached the
cap. That missed a window of empty logs. The function now returns
cap_reached when more_remain is true and the call hit the window LIMIT
or the batch budget. Unbounded (0) stays false.

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Cursor <cursoragent@cursor.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

Copilot code review Copilot
Copilot review effort
Applies to this pull request for everyone.Learn more about Copilot code review.
Copilot left review comments
@giuseppelillo giuseppelillo giuseppelillo approved these changes
@jeqo jeqo Awaiting requested review from jeqo

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@viktorsomogyi @giuseppelillo
0