diff --git a/force-app/main/default/classes/Async.cls b/force-app/main/default/classes/Async.cls
index f636834..3e214ff 100644
--- a/force-app/main/default/classes/Async.cls
+++ b/force-app/main/default/classes/Async.cls
@@ -7,6 +7,10 @@ public inherited sharing class Async {
         return new QueueableBuilder();
     }
 
+    public static ChunkBuilder chunk(ChunkJob job, ChunkSource source) {
+        return new ChunkBuilder(job, source);
+    }
+
     public static BatchableBuilder batchable(Object job) {
         return new BatchableBuilder(job);
     }
diff --git a/force-app/main/default/classes/AsyncTest.cls b/force-app/main/default/classes/AsyncTest.cls
index 447e80d..0aafb34 100644
--- a/force-app/main/default/classes/AsyncTest.cls
+++ b/force-app/main/default/classes/AsyncTest.cls
@@ -13,6 +13,7 @@ private class AsyncTest implements Database.Batchable<SObject> {
     private static final String TEST_SIGNATURE_NAME = 'SignatureName';
     private static final String TEST_SCHEDULABLE_JOB_NAME = 'SchedulableTestJob';
     private static final String PRIMITIVE_VALUE_INITIAL = 'INITIAL_VALUE';
+    private static final String CHUNK_PROCESSED = 'CHUNK_PROCESSED';
 
     @IsTest
     private static void shouldEnqueue60QueueablesSuccessfully() {
@@ -3320,4 +3321,508 @@ private class AsyncTest implements Database.Batchable<SObject> {
             insert new Account(Name = accountName, Description = 'Job: ' + jobId);
         }
     }
+
+    @IsTest
+    private static void shouldProcessFirstChunkThroughEnqueue() {
+        List<Account> accounts = createAccounts(5);
+
+        Test.startTest();
+        Async.Result result = Async.chunk(new MarkingChunkJob(), ChunkSource.of(accounts))
+            .chunkSize(5)
+            .enqueue();
+        Test.stopTest();
+
+        Assert.areNotEqual(null, result.salesforceJobId);
+        Assert.areEqual(
+            5,
+            [SELECT COUNT() FROM Account WHERE Site = :CHUNK_PROCESSED],
+            'The single chunk should process every record in the source.'
+        );
+    }
+
+    @IsTest
+    private static void shouldProcessAllInMemoryChunksAcrossPages() {
+        List<Account> accounts = createAccounts(6);
+
+        Test.startTest();
+        Async.chunk(new MarkingChunkJob(), ChunkSource.of(accounts)).chunkSize(3).enqueue();
+        Test.stopTest();
+
+        Assert.areEqual(
+            6,
+            [SELECT COUNT() FROM Account WHERE Site = :CHUNK_PROCESSED],
+            'Self-chaining must process every page of the run.'
+        );
+    }
+
+    @IsTest
+    private static void shouldProcessAllCursorChunksAcrossPages() {
+        createAccounts(6);
+
+        Test.startTest();
+        Async.chunk(new MarkingChunkJob(), ChunkSource.query('SELECT Id FROM Account'))
+            .chunkSize(3)
+            .enqueue();
+        Test.stopTest();
+
+        Assert.areEqual(
+            6,
+            [SELECT COUNT() FROM Account WHERE Site = :CHUNK_PROCESSED],
+            'A cursor must survive re-enqueue and fetch every page.'
+        );
+    }
+
+    @IsTest
+    private static void shouldCarryJobMemberStateAcrossChunks() {
+        List<Account> accounts = createAccounts(6);
+
+        Test.startTest();
+        Async.chunk(new LabelingChunkJob('RUN_LABEL'), ChunkSource.of(accounts))
+            .chunkSize(2)
+            .enqueue();
+        Test.stopTest();
+
+        Assert.areEqual(
+            6,
+            [SELECT COUNT() FROM Account WHERE Site = 'RUN_LABEL'],
+            'Job member state must survive cloning and serialization across every page.'
+        );
+    }
+
+    @IsTest
+    private static void shouldContinueToNextChunkAfterAFailedChunk() {
+        List<Account> accounts = new List<Account>{
+            new Account(Name = 'FAIL first'),
+            new Account(Name = 'ok second'),
+            new Account(Name = 'ok third'),
+            new Account(Name = 'ok fourth')
+        };
+        insert accounts;
+
+        Test.startTest();
+        Async.chunk(new FailMarkedChunkJob(), ChunkSource.of(accounts)).chunkSize(2).enqueue();
+        Test.stopTest();
+
+        Assert.areEqual(
+            2,
+            [SELECT COUNT() FROM Account WHERE Site = :CHUNK_PROCESSED],
+            'A failed chunk must not stop the remaining chunks by default.'
+        );
+    }
+
+    @IsTest
+    private static void shouldStopAfterFailedChunkWhenConfigured() {
+        List<Account> accounts = new List<Account>{
+            new Account(Name = 'FAIL first'),
+            new Account(Name = 'ok second'),
+            new Account(Name = 'ok third'),
+            new Account(Name = 'ok fourth')
+        };
+        insert accounts;
+
+        Test.startTest();
+        Async.chunk(new FailMarkedChunkJob(), ChunkSource.of(accounts))
+            .chunkSize(2)
+            .stopRemainingChunksOnFailure()
+            .enqueue();
+        Test.stopTest();
+
+        Assert.areEqual(
+            0,
+            [SELECT COUNT() FROM Account WHERE Site = :CHUNK_PROCESSED],
+            'stopRemainingChunksOnFailure must halt the run after a failed chunk.'
+        );
+    }
+
+    @IsTest
+    private static void shouldApplyAllChunkBuilderOptions() {
+        List<Account> accounts = createAccounts(2);
+
+        Test.startTest();
+        Async.Result result = Async.chunk(new MarkingChunkJob(), ChunkSource.of(accounts))
+            .chunkSize(2)
+            .priority(3)
+            .delay(1)
+            .retry(2)
+            .backoff(Backoff.fixed(1))
+            .retryOn(CustomException.class)
+            .mockId('chunk-mock')
+            .stopRemainingChunksOnFailure()
+            .enqueue();
+        Test.stopTest();
+
+        Assert.areNotEqual(
+            null,
+            result.salesforceJobId,
+            'Every builder option should still enqueue.'
+        );
+    }
+
+    @IsTest
+    private static void shouldProcessFirstChunkInBulk() {
+        List<Account> accounts = createAccounts(200);
+
+        Test.startTest();
+        Async.chunk(new MarkingChunkJob(), ChunkSource.of(accounts)).chunkSize(200).enqueue();
+        Test.stopTest();
+
+        Assert.areEqual(
+            200,
+            [SELECT COUNT() FROM Account WHERE Site = :CHUNK_PROCESSED],
+            'A 200-record chunk should process in one bulk-safe page.'
+        );
+    }
+
+    @IsTest
+    private static void shouldFetchCorrectOffsetForLaterChunk() {
+        List<Account> accounts = createAccounts(10);
+        MarkingChunkJob job = new MarkingChunkJob();
+        job.prepareRun(ChunkSource.of(accounts), 5, false, null, false);
+        job.position = 5;
+
+        job.work();
+
+        Set<Id> laterHalf = new Set<Id>();
+        for (Integer i = 5; i < 10; i++) {
+            laterHalf.add(accounts[i].Id);
+        }
+        List<Account> processed = [
+            SELECT Id
+            FROM Account
+            WHERE Site = :CHUNK_PROCESSED
+        ];
+        Assert.areEqual(
+            5,
+            processed.size(),
+            'Only the second page of records should be processed.'
+        );
+        for (Account processedAccount : processed) {
+            Assert.isTrue(
+                laterHalf.contains(processedAccount.Id),
+                'Offset fetch must return records from position 5 onward.'
+            );
+        }
+    }
+
+    @IsTest
+    private static void shouldCreateResultRowForChunk() {
+        List<Account> accounts = createAccounts(3);
+        QueueableChain chain = new QueueableChain();
+        chain.queueableJobSettingByJobName = new Map<String, QueueableJobSetting__mdt>{
+            QueueableManager.QUEUEABLE_JOB_SETTING_ALL => new QueueableJobSetting__mdt(
+                DeveloperName = QueueableManager.QUEUEABLE_JOB_SETTING_ALL,
+                CreateResult__c = true
+            )
+        };
+
+        Test.startTest();
+        QueueableManager.get().setChain(chain);
+        Async.chunk(new MarkingChunkJob(), ChunkSource.of(accounts)).chunkSize(3).enqueue();
+        Test.stopTest();
+
+        List<AsyncResult__c> results = [
+            SELECT Id, Status__c, ChainId__c, ClassName__c
+            FROM AsyncResult__c
+        ];
+        Assert.areEqual(1, results.size(), 'A completed chunk page should record one AsyncResult.');
+        Assert.areEqual(QueueableManager.STATUS_COMPLETED, results[0].Status__c);
+        Assert.areNotEqual(null, results[0].ChainId__c);
+    }
+
+    @IsTest
+    private static void shouldReturnNextChunkWhenRecordsRemain() {
+        MarkingChunkJob job = new MarkingChunkJob();
+        job.prepareRun(ChunkSource.of(createAccounts(10)), 5, false, null, false);
+
+        ChunkJob nextPage = job.nextPageOrNull();
+
+        Assert.areNotEqual(null, nextPage, 'A next page is expected while records remain.');
+        Assert.areEqual(5, nextPage.position, 'The next page must advance by chunkSize.');
+    }
+
+    @IsTest
+    private static void shouldNotCarryInitialDelayToLaterChunks() {
+        MarkingChunkJob job = new MarkingChunkJob();
+        job.prepareRun(ChunkSource.of(createAccounts(10)), 5, false, null, false);
+        job.delay = 5;
+
+        ChunkJob nextPage = job.nextPageOrNull();
+
+        Assert.areEqual(null, nextPage.delay, 'An initial delay must not repeat on every page.');
+    }
+
+    @IsTest
+    private static void shouldApplyDelayBetweenChunks() {
+        MarkingChunkJob job = new MarkingChunkJob();
+        job.prepareRun(ChunkSource.of(createAccounts(10)), 5, false, 3, false);
+
+        ChunkJob nextPage = job.nextPageOrNull();
+
+        Assert.areEqual(3, nextPage.delay, 'delayBetweenChunks must throttle each later page.');
+    }
+
+    @IsTest
+    private static void shouldRejectDeepCloneChunkJob() {
+        MarkingChunkJob job = new MarkingChunkJob();
+        job.deepClone = true;
+
+        try {
+            Async.chunk(job, ChunkSource.of(createAccounts(2)));
+            Assert.fail('A deepClone ChunkJob should be rejected at build time.');
+        } catch (Exception ex) {
+            Assert.areEqual(ChunkBuilder.ERROR_MESSAGE_DEEP_CLONE_UNSUPPORTED, ex.getMessage());
+        }
+    }
+
+    @IsTest
+    private static void shouldPruneSettledPagesByDefault() {
+        MarkingChunkJob job = new MarkingChunkJob();
+        job.prepareRun(ChunkSource.of(createAccounts(4)), 2, false, null, false);
+
+        Assert.isFalse(job.shouldKeepPages(), 'Settled pages are pruned by default.');
+    }
+
+    @IsTest
+    private static void shouldKeepSettledPagesWhenRequested() {
+        List<Account> accounts = createAccounts(6);
+
+        Test.startTest();
+        Async.chunk(new MarkingChunkJob(), ChunkSource.of(accounts))
+            .chunkSize(2)
+            .keepChunkPages()
+            .enqueue();
+        Test.stopTest();
+
+        Assert.areEqual(
+            6,
+            [SELECT COUNT() FROM Account WHERE Site = :CHUNK_PROCESSED],
+            'A kept-pages run still processes every page.'
+        );
+    }
+
+    @IsTest
+    private static void shouldReturnNoNextChunkAtEndOfSource() {
+        MarkingChunkJob job = new MarkingChunkJob();
+        job.prepareRun(ChunkSource.of(createAccounts(10)), 5, false, null, false);
+        job.position = 5;
+
+        Assert.areEqual(null, job.nextPageOrNull(), 'No page should follow the final chunk.');
+    }
+
+    @IsTest
+    private static void shouldStopRemainingChunksOnFailureWhenConfigured() {
+        MarkingChunkJob stopping = new MarkingChunkJob();
+        stopping.prepareRun(ChunkSource.of(createAccounts(10)), 5, true, null, false);
+        stopping.hasFailed = true;
+
+        Assert.areEqual(
+            null,
+            stopping.nextPageOrNull(),
+            'A failed chunk must halt the run when stopRemainingChunksOnFailure is set.'
+        );
+    }
+
+    @IsTest
+    private static void shouldContinueRemainingChunksOnFailureByDefault() {
+        MarkingChunkJob continuing = new MarkingChunkJob();
+        continuing.prepareRun(ChunkSource.of(createAccounts(10)), 5, false, null, false);
+        continuing.hasFailed = true;
+
+        Assert.areNotEqual(
+            null,
+            continuing.nextPageOrNull(),
+            'A failed chunk must not halt the run by default.'
+        );
+    }
+
+    @IsTest
+    private static void shouldTreatEmptySourceAsNoOp() {
+        Async.Result result = Async.chunk(
+                new MarkingChunkJob(),
+                ChunkSource.of(new List<Account>())
+            )
+            .enqueue();
+
+        Assert.areEqual(null, result.salesforceJobId, 'An empty source should enqueue nothing.');
+        Assert.isTrue(result.queueableChainState.jobs.isEmpty());
+    }
+
+    @IsTest
+    private static void shouldRejectInvalidChunkSize() {
+        ChunkBuilder builder = Async.chunk(
+            new MarkingChunkJob(),
+            ChunkSource.of(createAccounts(1))
+        );
+        for (Integer invalid : new List<Integer>{ 0, -1 }) {
+            try {
+                builder.chunkSize(invalid);
+                Assert.fail('chunkSize ' + invalid + ' should be rejected.');
+            } catch (Exception ex) {
+                Assert.areEqual(ChunkBuilder.ERROR_MESSAGE_INVALID_CHUNK_SIZE, ex.getMessage());
+            }
+        }
+    }
+
+    @IsTest
+    private static void shouldRejectNullJobAndSource() {
+        try {
+            Async.chunk(null, ChunkSource.of(createAccounts(1)));
+            Assert.fail('A null job should be rejected.');
+        } catch (Exception ex) {
+            Assert.areEqual(ChunkBuilder.ERROR_MESSAGE_NULL_JOB, ex.getMessage());
+        }
+        try {
+            Async.chunk(new MarkingChunkJob(), null);
+            Assert.fail('A null source should be rejected.');
+        } catch (Exception ex) {
+            Assert.areEqual(ChunkBuilder.ERROR_MESSAGE_NULL_SOURCE, ex.getMessage());
+        }
+    }
+
+    @IsTest
+    private static void shouldRejectRetryAboveCapOnChunk() {
+        ChunkBuilder builder = Async.chunk(
+            new MarkingChunkJob(),
+            ChunkSource.of(createAccounts(1))
+        );
+        try {
+            builder.retry(QueueableManager.MAX_RETRY_CAP + 1);
+            Assert.fail('Retry above the cap should be rejected.');
+        } catch (Exception ex) {
+            Assert.areEqual(
+                QueueableManager.ERROR_MESSAGE_MAX_RETRIES_EXCEEDS_CAP,
+                ex.getMessage()
+            );
+        }
+    }
+
+    @IsTest
+    private static void shouldPluckIdsForIdChunkJob() {
+        List<Account> accounts = createAccounts(5);
+        CapturingIdChunkJob job = new CapturingIdChunkJob();
+
+        job.work(accounts);
+
+        Assert.areEqual(5, job.capturedIds.size(), 'Every record id should be plucked.');
+        for (Account account : accounts) {
+            Assert.isTrue(job.capturedIds.contains(account.Id));
+        }
+    }
+
+    @IsTest
+    private static void shouldExposeInMemorySourceRecordsAndSize() {
+        List<Account> accounts = createAccounts(5);
+        ChunkSource source = ChunkSource.of(accounts);
+
+        Assert.areEqual(5, source.size());
+        Assert.areEqual(2, source.fetch(0, 2).size());
+        Assert.areEqual(
+            1,
+            source.fetch(4, 5).size(),
+            'A short final page must clamp to the remainder.'
+        );
+    }
+
+    @IsTest
+    private static void shouldBuildIdSourceRecordsWithIds() {
+        List<Account> accounts = createAccounts(3);
+        Set<Id> ids = new Map<Id, Account>(accounts).keySet();
+        ChunkSource source = ChunkSource.ofIds(ids);
+
+        Assert.areEqual(3, source.size());
+        Set<Id> fetched = new Set<Id>();
+        for (SObject record : source.fetch(0, 3)) {
+            fetched.add(record.Id);
+        }
+        Assert.areEqual(ids, fetched);
+    }
+
+    @IsTest
+    private static void shouldDelegateToCursorSource() {
+        createAccounts(5);
+        ChunkSource source = ChunkSource.cursor(Database.getCursor('SELECT Id FROM Account'));
+
+        Assert.areEqual(5, source.size());
+        Assert.areEqual(2, source.fetch(0, 2).size());
+    }
+
+    @IsTest
+    private static void shouldOpenCursorFromQuery() {
+        createAccounts(4);
+        ChunkSource source = ChunkSource.query('SELECT Id FROM Account');
+
+        Assert.areEqual(4, source.size());
+    }
+
+    @IsTest
+    private static void shouldRejectNullCursorAndBlankQuery() {
+        try {
+            ChunkSource.cursor(null);
+            Assert.fail('A null cursor should be rejected.');
+        } catch (Exception ex) {
+            Assert.areEqual(ChunkSource.ERROR_MESSAGE_NULL_CURSOR, ex.getMessage());
+        }
+        try {
+            ChunkSource.query('   ');
+            Assert.fail('A blank query should be rejected.');
+        } catch (Exception ex) {
+            Assert.areEqual(ChunkSource.ERROR_MESSAGE_BLANK_QUERY, ex.getMessage());
+        }
+    }
+
+    private static List<Account> createAccounts(Integer count) {
+        List<Account> accounts = new List<Account>();
+        for (Integer i = 0; i < count; i++) {
+            accounts.add(new Account(Name = 'Chunk ' + i));
+        }
+        insert accounts;
+        return accounts;
+    }
+
+    private class MarkingChunkJob extends ChunkJob {
+        public override void work(List<SObject> chunk) {
+            List<Account> toUpdate = new List<Account>();
+            for (SObject record : chunk) {
+                toUpdate.add(new Account(Id = record.Id, Site = CHUNK_PROCESSED));
+            }
+            update toUpdate;
+        }
+    }
+
+    private class LabelingChunkJob extends ChunkJob {
+        private String label;
+
+        public LabelingChunkJob(String label) {
+            this.label = label;
+        }
+
+        public override void work(List<SObject> chunk) {
+            List<Account> toUpdate = new List<Account>();
+            for (SObject record : chunk) {
+                toUpdate.add(new Account(Id = record.Id, Site = label));
+            }
+            update toUpdate;
+        }
+    }
+
+    private class FailMarkedChunkJob extends ChunkJob {
+        public override void work(List<SObject> chunk) {
+            List<Account> toUpdate = new List<Account>();
+            for (SObject record : chunk) {
+                Account account = (Account) record;
+                if (account.Name != null && account.Name.contains('FAIL')) {
+                    throw new CustomException(AsyncTest.CUSTOM_ERROR_MESSAGE);
+                }
+                toUpdate.add(new Account(Id = account.Id, Site = CHUNK_PROCESSED));
+            }
+            update toUpdate;
+        }
+    }
+
+    private class CapturingIdChunkJob extends IdChunkJob {
+        public Set<Id> capturedIds;
+        public override void work(Set<Id> ids) {
+            this.capturedIds = ids;
+        }
+    }
 }
diff --git a/force-app/main/default/classes/queue/ChunkBuilder.cls b/force-app/main/default/classes/queue/ChunkBuilder.cls
new file mode 100644
index 0000000..785d98e
--- /dev/null
+++ b/force-app/main/default/classes/queue/ChunkBuilder.cls
@@ -0,0 +1,109 @@
+public inherited sharing class ChunkBuilder {
+    @TestVisible
+    private static final String ERROR_MESSAGE_INVALID_CHUNK_SIZE = 'chunkSize must be a positive integer';
+    @TestVisible
+    private static final String ERROR_MESSAGE_NULL_JOB = 'ChunkJob cannot be null';
+    @TestVisible
+    private static final String ERROR_MESSAGE_NULL_SOURCE = 'ChunkSource cannot be null';
+    @TestVisible
+    private static final String ERROR_MESSAGE_DEEP_CLONE_UNSUPPORTED = 'deepClone is not supported for a ChunkJob: a Database.Cursor source cannot be JSON-serialized.';
+    private static final Integer DEFAULT_CHUNK_SIZE = 200;
+
+    private ChunkJob job;
+    private ChunkSource source;
+    private Integer chunkSize = DEFAULT_CHUNK_SIZE;
+    private Boolean stopRemainingChunksOnFailure = false;
+    private Integer delayBetweenChunks;
+    private Boolean keepPages = false;
+
+    public ChunkBuilder(ChunkJob job, ChunkSource source) {
+        if (job == null) {
+            throw new IllegalArgumentException(ERROR_MESSAGE_NULL_JOB);
+        }
+        if (source == null) {
+            throw new IllegalArgumentException(ERROR_MESSAGE_NULL_SOURCE);
+        }
+        if (job.deepClone) {
+            throw new IllegalArgumentException(ERROR_MESSAGE_DEEP_CLONE_UNSUPPORTED);
+        }
+        this.job = (ChunkJob) job.clone();
+        this.source = source;
+    }
+
+    public ChunkBuilder chunkSize(Integer chunkSize) {
+        if (chunkSize == null || chunkSize <= 0) {
+            throw new IllegalArgumentException(ERROR_MESSAGE_INVALID_CHUNK_SIZE);
+        }
+        this.chunkSize = chunkSize;
+        return this;
+    }
+
+    public ChunkBuilder stopRemainingChunksOnFailure() {
+        this.stopRemainingChunksOnFailure = true;
+        return this;
+    }
+
+    public ChunkBuilder priority(Integer priority) {
+        job.priority = priority;
+        return this;
+    }
+
+    public ChunkBuilder delay(Integer delay) {
+        job.delay = delay;
+        return this;
+    }
+
+    public ChunkBuilder delayBetweenChunks(Integer minutes) {
+        this.delayBetweenChunks = minutes;
+        return this;
+    }
+
+    // Retain settled pages in the chain for introspection instead of pruning them. Off by
+    // default; not for large runs, since kept pages grow the serialized chain state.
+    public ChunkBuilder keepChunkPages() {
+        this.keepPages = true;
+        return this;
+    }
+
+    public ChunkBuilder retry(Integer maxRetries) {
+        QueueableManager.assertValidMaxRetries(maxRetries);
+        job.maxRetries = maxRetries;
+        return this;
+    }
+
+    public ChunkBuilder backoff(Backoff backoff) {
+        job.backoff = backoff;
+        return this;
+    }
+
+    public ChunkBuilder retryOn(Type exceptionType) {
+        return retryOn(new List<Type>{ exceptionType });
+    }
+
+    public ChunkBuilder retryOn(List<Type> exceptionTypes) {
+        QueueableManager.addRetryOnTypes(job, exceptionTypes);
+        return this;
+    }
+
+    public ChunkBuilder mockId(String mockId) {
+        job.mockId = mockId;
+        return this;
+    }
+
+    public Async.Result enqueue() {
+        if (source.size() == 0) {
+            return new Async.Result((Id) null)
+                .setQueueableChainState(
+                    new Async.QueueableChainState().setJobs(new List<QueueableJob>())
+                );
+        }
+        job.prepareRun(
+            source,
+            chunkSize,
+            stopRemainingChunksOnFailure,
+            delayBetweenChunks,
+            keepPages
+        );
+        return QueueableManager.get().chainAndEnqueue(job);
+    }
+}
diff --git a/force-app/main/default/classes/queue/ChunkBuilder.cls-meta.xml b/force-app/main/default/classes/queue/ChunkBuilder.cls-meta.xml
new file mode 100644
index 0000000..cad713d
--- /dev/null
+++ b/force-app/main/default/classes/queue/ChunkBuilder.cls-meta.xml
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<ApexClass xmlns="http://soap.sforce.com/2006/04/metadata">
+    <apiVersion>66.0</apiVersion>
+    <status>Active</status>
+</ApexClass>
diff --git a/force-app/main/default/classes/queue/ChunkJob.cls b/force-app/main/default/classes/queue/ChunkJob.cls
new file mode 100644
index 0000000..7421155
--- /dev/null
+++ b/force-app/main/default/classes/queue/ChunkJob.cls
@@ -0,0 +1,67 @@
+public inherited sharing abstract class ChunkJob extends QueueableJob {
+    @TestVisible
+    private ChunkSource source;
+    @TestVisible
+    private Integer chunkSize;
+    @TestVisible
+    private Integer position = 0;
+    @TestVisible
+    private Integer totalSize;
+    @TestVisible
+    private Boolean stopRemainingChunksOnFailure = false;
+    @TestVisible
+    private Integer delayBetweenChunks;
+    @TestVisible
+    private Boolean keepPages = false;
+
+    public abstract void work(List<SObject> chunk);
+
+    public override void work() {
+        Integer count = Math.min(chunkSize, totalSize - position);
+        work(source.fetch(position, count));
+    }
+
+    public void prepareRun(
+        ChunkSource source,
+        Integer chunkSize,
+        Boolean stopRemainingChunksOnFailure,
+        Integer delayBetweenChunks,
+        Boolean keepPages
+    ) {
+        this.source = source;
+        this.chunkSize = chunkSize;
+        this.stopRemainingChunksOnFailure = stopRemainingChunksOnFailure;
+        this.delayBetweenChunks = delayBetweenChunks;
+        this.keepPages = keepPages;
+        this.position = 0;
+        this.totalSize = source.size();
+        // Chunk errors are recorded in AsyncResult, not rethrown, so one bad page never aborts the run.
+        this.continueOnJobExecuteFail = true;
+    }
+
+    public Boolean shouldKeepPages() {
+        return keepPages;
+    }
+
+    public ChunkJob nextPageOrNull() {
+        if (hasFailed && stopRemainingChunksOnFailure) {
+            return null;
+        }
+        Integer nextPosition = position + chunkSize;
+        if (nextPosition >= totalSize) {
+            return null;
+        }
+        ChunkJob nextPage = (ChunkJob) this.cloneJob();
+        nextPage.position = nextPosition;
+        nextPage.delay = delayBetweenChunks;
+        nextPage.retryAttempt = 0;
+        nextPage.hasFailed = false;
+        nextPage.failure = null;
+        nextPage.retryDecision = false;
+        nextPage.retryHistory = null;
+        nextPage.finalizerCtx = null;
+        nextPage.isProcessed = false;
+        nextPage.resultCreated = false;
+        return nextPage;
+    }
+}
diff --git a/force-app/main/default/classes/queue/ChunkJob.cls-meta.xml b/force-app/main/default/classes/queue/ChunkJob.cls-meta.xml
new file mode 100644
index 0000000..cad713d
--- /dev/null
+++ b/force-app/main/default/classes/queue/ChunkJob.cls-meta.xml
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<ApexClass xmlns="http://soap.sforce.com/2006/04/metadata">
+    <apiVersion>66.0</apiVersion>
+    <status>Active</status>
+</ApexClass>
diff --git a/force-app/main/default/classes/queue/ChunkSource.cls b/force-app/main/default/classes/queue/ChunkSource.cls
new file mode 100644
index 0000000..fa6d851
--- /dev/null
+++ b/force-app/main/default/classes/queue/ChunkSource.cls
@@ -0,0 +1,72 @@
+public inherited sharing abstract class ChunkSource {
+    public static final String ERROR_MESSAGE_NULL_CURSOR = 'Database.Cursor cannot be null';
+    public static final String ERROR_MESSAGE_BLANK_QUERY = 'SOQL query cannot be blank';
+
+    public abstract Integer size();
+    public abstract List<SObject> fetch(Integer offset, Integer count);
+
+    public static ChunkSource of(List<SObject> records) {
+        return new InMemoryChunkSource(records);
+    }
+
+    public static ChunkSource ofIds(Set<Id> ids) {
+        List<SObject> records = new List<SObject>();
+        if (ids != null) {
+            for (Id recordId : ids) {
+                records.add(recordId.getSObjectType().newSObject(recordId));
+            }
+        }
+        return new InMemoryChunkSource(records);
+    }
+
+    public static ChunkSource cursor(Database.Cursor cursor) {
+        if (cursor == null) {
+            throw new IllegalArgumentException(ERROR_MESSAGE_NULL_CURSOR);
+        }
+        return new CursorChunkSource(cursor);
+    }
+
+    public static ChunkSource query(String soql) {
+        if (String.isBlank(soql)) {
+            throw new IllegalArgumentException(ERROR_MESSAGE_BLANK_QUERY);
+        }
+        return new CursorChunkSource(Database.getCursor(soql));
+    }
+
+    public class InMemoryChunkSource extends ChunkSource {
+        private List<SObject> records;
+
+        private InMemoryChunkSource(List<SObject> records) {
+            this.records = records == null ? new List<SObject>() : records;
+        }
+
+        public override Integer size() {
+            return records.size();
+        }
+
+        public override List<SObject> fetch(Integer offset, Integer count) {
+            List<SObject> page = new List<SObject>();
+            Integer endIndex = Math.min(offset + count, records.size());
+            for (Integer i = offset; i < endIndex; i++) {
+                page.add(records[i]);
+            }
+            return page;
+        }
+    }
+
+    public class CursorChunkSource extends ChunkSource {
+        private Database.Cursor cursor;
+
+        private CursorChunkSource(Database.Cursor cursor) {
+            this.cursor = cursor;
+        }
+
+        public override Integer size() {
+            return cursor.getNumRecords();
+        }
+
+        public override List<SObject> fetch(Integer offset, Integer count) {
+            return cursor.fetch(offset, count);
+        }
+    }
+}
diff --git a/force-app/main/default/classes/queue/ChunkSource.cls-meta.xml b/force-app/main/default/classes/queue/ChunkSource.cls-meta.xml
new file mode 100644
index 0000000..cad713d
--- /dev/null
+++ b/force-app/main/default/classes/queue/ChunkSource.cls-meta.xml
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<ApexClass xmlns="http://soap.sforce.com/2006/04/metadata">
+    <apiVersion>66.0</apiVersion>
+    <status>Active</status>
+</ApexClass>
diff --git a/force-app/main/default/classes/queue/IdChunkJob.cls b/force-app/main/default/classes/queue/IdChunkJob.cls
new file mode 100644
index 0000000..8a51caf
--- /dev/null
+++ b/force-app/main/default/classes/queue/IdChunkJob.cls
@@ -0,0 +1,13 @@
+public inherited sharing abstract class IdChunkJob extends ChunkJob {
+    public abstract void work(Set<Id> ids);
+
+    public override void work(List<SObject> chunk) {
+        Set<Id> ids = new Set<Id>();
+        for (SObject record : chunk) {
+            if (record.Id != null) {
+                ids.add(record.Id);
+            }
+        }
+        work(ids);
+    }
+}
diff --git a/force-app/main/default/classes/queue/IdChunkJob.cls-meta.xml b/force-app/main/default/classes/queue/IdChunkJob.cls-meta.xml
new file mode 100644
index 0000000..cad713d
--- /dev/null
+++ b/force-app/main/default/classes/queue/IdChunkJob.cls-meta.xml
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<ApexClass xmlns="http://soap.sforce.com/2006/04/metadata">
+    <apiVersion>66.0</apiVersion>
+    <status>Active</status>
+</ApexClass>
diff --git a/force-app/main/default/classes/queue/QueueableBuilder.cls b/force-app/main/default/classes/queue/QueueableBuilder.cls
index b26dac8..73f080c 100644
--- a/force-app/main/default/classes/queue/QueueableBuilder.cls
+++ b/force-app/main/default/classes/queue/QueueableBuilder.cls
@@ -76,14 +76,7 @@ public inherited sharing class QueueableBuilder {
     }
 
     public QueueableBuilder retry(Integer maxRetries) {
-        if (maxRetries == null || maxRetries < 0) {
-            throw new IllegalArgumentException(QueueableManager.ERROR_MESSAGE_INVALID_MAX_RETRIES);
-        }
-        if (maxRetries > QueueableManager.MAX_RETRY_CAP) {
-            throw new IllegalArgumentException(
-                QueueableManager.ERROR_MESSAGE_MAX_RETRIES_EXCEEDS_CAP
-            );
-        }
+        QueueableManager.assertValidMaxRetries(maxRetries);
         job.maxRetries = maxRetries;
         return this;
     }
@@ -98,14 +91,7 @@ public inherited sharing class QueueableBuilder {
     }
 
     public QueueableBuilder retryOn(List<Type> exceptionTypes) {
-        if (job.retryOnExceptionTypes == null) {
-            job.retryOnExceptionTypes = new Set<String>();
-        }
-        for (Type exceptionType : exceptionTypes) {
-            if (exceptionType != null) {
-                job.retryOnExceptionTypes.add(exceptionType.getName());
-            }
-        }
+        QueueableManager.addRetryOnTypes(job, exceptionTypes);
         return this;
     }
 
diff --git a/force-app/main/default/classes/queue/QueueableChain.cls b/force-app/main/default/classes/queue/QueueableChain.cls
index 37f63d1..c6d4fb1 100644
--- a/force-app/main/default/classes/queue/QueueableChain.cls
+++ b/force-app/main/default/classes/queue/QueueableChain.cls
@@ -89,9 +89,37 @@ public inherited sharing class QueueableChain {
         recordOutcome(previousJob);
 
         setFinalizerContextToAllFinalizersForPreviousJob(previousJob, ctx);
+        maybeAdvanceChunkRun(previousJob);
         enqueueNextJobIfAny();
     }
 
+    private void maybeAdvanceChunkRun(QueueableJob previousJob) {
+        if (!(previousJob instanceof ChunkJob)) {
+            return;
+        }
+        ChunkJob settledPage = (ChunkJob) previousJob;
+        ChunkJob nextPage = settledPage.nextPageOrNull();
+        if (nextPage != null) {
+            addJob(nextPage);
+        }
+        if (!settledPage.shouldKeepPages()) {
+            removeRecordedChunkPages();
+        }
+    }
+
+    // A chunk run appends pages one at a time; dropping recorded pages keeps the serialized
+    // chain state flat instead of piling up every page of the run.
+    private void removeRecordedChunkPages() {
+        for (Integer i = jobs.size() - 1; i >= 0; i--) {
+            QueueableJob job = jobs[i];
+            if (job instanceof ChunkJob && job.isProcessed && job.resultCreated) {
+                outcomeByCustomJobId.remove(job.customJobId);
+                resultIdByCustomJobId.remove(job.customJobId);
+                jobs.remove(i);
+            }
+        }
+    }
+
     private void reconcileOutcomeFromFinalizer(QueueableJob job, FinalizerContext ctx) {
         if (ctx?.getResult() != ParentJobResult.UNHANDLED_EXCEPTION || job.hasFailed) {
             return;
diff --git a/force-app/main/default/classes/queue/QueueableManager.cls b/force-app/main/default/classes/queue/QueueableManager.cls
index 378f4a4..a0189f7 100644
--- a/force-app/main/default/classes/queue/QueueableManager.cls
+++ b/force-app/main/default/classes/queue/QueueableManager.cls
@@ -135,6 +135,26 @@ public inherited sharing class QueueableManager {
             (Limits.getQueueableJobs() >= Limits.getLimitQueueableJobs() || System.isQueueable());
     }
 
+    public static void assertValidMaxRetries(Integer maxRetries) {
+        if (maxRetries == null || maxRetries < 0) {
+            throw new IllegalArgumentException(ERROR_MESSAGE_INVALID_MAX_RETRIES);
+        }
+        if (maxRetries > MAX_RETRY_CAP) {
+            throw new IllegalArgumentException(ERROR_MESSAGE_MAX_RETRIES_EXCEEDS_CAP);
+        }
+    }
+
+    public static void addRetryOnTypes(QueueableJob job, List<Type> exceptionTypes) {
+        if (job.retryOnExceptionTypes == null) {
+            job.retryOnExceptionTypes = new Set<String>();
+        }
+        for (Type exceptionType : exceptionTypes) {
+            if (exceptionType != null) {
+                job.retryOnExceptionTypes.add(exceptionType.getName());
+            }
+        }
+    }
+
     public static QueueableManager get() {
         if (instance == null) {
             instance = new QueueableManager();
diff --git a/scripts/create-unlocked-package-version.sh b/scripts/create-unlocked-package-version.sh
index 4edc6f6..0b9e9ea 100755
--- a/scripts/create-unlocked-package-version.sh
+++ b/scripts/create-unlocked-package-version.sh
@@ -24,6 +24,10 @@ TARGET_FILES=(
     "force-app/main/default/classes/Async.cls"
     "force-app/main/default/classes/queue/QueueableJob.cls"
     "force-app/main/default/classes/queue/QueueableBuilder.cls"
+    "force-app/main/default/classes/queue/ChunkSource.cls"
+    "force-app/main/default/classes/queue/ChunkJob.cls"
+    "force-app/main/default/classes/queue/IdChunkJob.cls"
+    "force-app/main/default/classes/queue/ChunkBuilder.cls"
     "force-app/main/default/classes/queue/Backoff.cls"
     "force-app/main/default/classes/batch/BatchableBuilder.cls"
     "force-app/main/default/classes/cleanup/AsyncResultCleanupBatch.cls"
@@ -53,6 +57,15 @@ sedi 's/global QueueableChainState setEnqueueType/public QueueableChainState set
 sedi 's/global void enqueue(QueueableChain chain)/public void enqueue(QueueableChain chain)/g' \
     "force-app/main/default/classes/queue/QueueableJob.cls"
 
+# prepareRun and nextPageOrNull are chunk-run wiring (ChunkBuilder sets up the run, QueueableChain
+# drives paging); they are not part of the consumer API.
+sedi 's/global void prepareRun(/public void prepareRun(/g' \
+    "force-app/main/default/classes/queue/ChunkJob.cls"
+sedi 's/global ChunkJob nextPageOrNull(/public ChunkJob nextPageOrNull(/g' \
+    "force-app/main/default/classes/queue/ChunkJob.cls"
+sedi 's/global Boolean shouldKeepPages(/public Boolean shouldKeepPages(/g' \
+    "force-app/main/default/classes/queue/ChunkJob.cls"
+
 # The blanket rename rewrites the "public override ..." guidance string inside
 # QueueableJob.cloneJob(); restore it so consumers get correct override syntax.
 sedi 's/global override QueueableJob cloneForDeepCopy/public override QueueableJob cloneForDeepCopy/g' \
diff --git a/website/.vitepress/config.mts b/website/.vitepress/config.mts
index ca81b63..dbce1a2 100644
--- a/website/.vitepress/config.mts
+++ b/website/.vitepress/config.mts
@@ -55,6 +55,7 @@ export default defineConfig({
         collapsed: false,
         items: [
           { text: 'Queueable', link: '/api/queueable' },
+          { text: 'Chunk', link: '/api/chunk' },
           { text: 'Batchable', link: '/api/batchable' },
           { text: 'Schedulable', link: '/api/schedulable' },
           { text: 'AsyncMock', link: '/api/async-mock' }
diff --git a/website/api/chunk.md b/website/api/chunk.md
new file mode 100644
index 0000000..42630f1
--- /dev/null
+++ b/website/api/chunk.md
@@ -0,0 +1,284 @@
+# Chunk API
+
+Apex classes `ChunkBuilder.cls`, `ChunkJob.cls`, `IdChunkJob.cls`, and
+`ChunkSource.cls`.
+
+Process a large data set in tuned pages across chained Queueables, with the same
+per-job tracking and retry/backoff as a normal `QueueableJob`. Reach for this when
+you need to process thousands of records in the background but do not want a
+`Database.Batchable` (heavier, separate lifecycle) or one job per record.
+
+Each page runs as its own tracked job in the chain. When a page settles, the
+framework appends the next page and drops the settled one, so only the current
+position moves forward and a huge run never materializes a giant job list.
+
+**Common ChunkJob class example:**
+
+Extend `ChunkJob` and put your per-page logic in `work(List<SObject> chunk)`.
+
+```apex
+public class AccountRecalcJob extends ChunkJob {
+  public override void work(List<SObject> chunk) {
+    List<Account> accounts = (List<Account>) chunk;
+    for (Account acc : accounts) {
+      acc.Description = 'Recalculated';
+    }
+    update accounts;
+  }
+}
+```
+
+```apex
+Async.Result result = Async.chunk(new AccountRecalcJob(), ChunkSource.of(records))
+	.chunkSize(50)
+	.priority(5)
+	.retry(2)
+	.enqueue();
+```
+
+::: warning deepClone is not supported for a ChunkJob
+Pages share one source and are isolated by serialization at each enqueue, and a
+`Database.Cursor` source cannot be JSON-serialized. `Async.chunk(...)` throws if
+the job has `deepClone` set. Build the job fully before handing it to
+`Async.chunk(...)` (do not mutate its members afterward).
+:::
+
+**Common IdChunkJob class example:**
+
+When you only need the record ids, extend `IdChunkJob` and override
+`work(Set<Id> ids)`. The framework plucks the ids from each fetched page.
+
+```apex
+public class AccountCloseJob extends IdChunkJob {
+  public override void work(Set<Id> ids) {
+    // requery or process by id
+  }
+}
+```
+
+**Large set via SOQL cursor example:**
+
+For sets too large to hold in memory, hand the framework a `Database.Cursor` (or a
+SOQL string it opens for you). The cursor is never fully materialized; each page
+fetches its slice.
+
+```apex
+Async.chunk(new AccountRecalcJob(), ChunkSource.query('SELECT Id FROM Account WHERE ...'))
+	.chunkSize(200)
+	.enqueue();
+```
+
+## Failure handling
+
+Two independent levels:
+
+- **Per chunk (retry).** A page that throws retries per `.retry(...)` /
+  `.backoff(...)`, exactly like a `QueueableJob`. Each terminal page records an
+  `AsyncResult__c` row (`COMPLETED` or `FAILED`).
+- **Across the run.** By default the run keeps processing the remaining chunks
+  even if one fails (best-effort, `Batchable`-like). Call
+  `.stopRemainingChunksOnFailure()` to stop the run when a chunk exhausts its
+  retries.
+
+## Cursor governor caps
+
+`ChunkSource.cursor(...)` / `ChunkSource.query(...)` open a SOQL cursor. Salesforce
+enforces cursor limits you should weigh against a `Database.Batchable`:
+
+| Limit                                    | Value          |
+| ---------------------------------------- | -------------- |
+| Rows per cursor                          | 50 million     |
+| Cursor fetch calls per day               | see Salesforce cursor limits |
+| Cursor lifespan                          | 2 days         |
+| Concurrent cursors                       | see Salesforce cursor limits |
+
+`ChunkSource.of(...)` / `ChunkSource.ofIds(...)` hold records in memory instead, so
+they are bounded by heap, not cursor limits. Prefer them for smaller sets and for
+tests.
+
+## Security
+
+`ChunkSource.query(String soql)` runs the SOQL you pass, so you own its safety.
+Bind user input, and add `WITH USER_MODE` to the query if the run must respect the
+current user's field and record access. The framework hands each page to your
+`work(...)` as queried; it does not strip fields.
+
+## Methods
+
+### INIT
+
+#### chunk
+
+Constructs a new `ChunkBuilder` for the given job and record source. The source is
+mandatory.
+
+**Signature**
+
+```apex
+ChunkBuilder chunk(ChunkJob job, ChunkSource source);
+```
+
+**Example**
+
+```apex
+Async.chunk(new AccountRecalcJob(), ChunkSource.of(records));
+```
+
+### Source
+
+`ChunkSource` is the pluggable seam between the framework and your data. Use a
+factory, or extend `ChunkSource` for a fully custom source.
+
+| Factory                             | Source                                       |
+| ----------------------------------- | -------------------------------------------- |
+| `ChunkSource.of(List<SObject>)`     | in-memory records (also the test fake)       |
+| `ChunkSource.ofIds(Set<Id>)`        | id-only source                               |
+| `ChunkSource.cursor(Database.Cursor)` | wraps an existing SOQL cursor              |
+| `ChunkSource.query(String soql)`    | opens a cursor over the query                |
+
+**Signature**
+
+```apex
+abstract Integer size();
+abstract List<SObject> fetch(Integer offset, Integer count);
+```
+
+### Build
+
+#### chunkSize
+
+Sets how many records each page processes. Must be a positive integer. Defaults to
+`200`.
+
+**Signature**
+
+```apex
+ChunkBuilder chunkSize(Integer chunkSize);
+```
+
+**Example**
+
+```apex
+Async.chunk(new AccountRecalcJob(), ChunkSource.of(records)).chunkSize(50);
+```
+
+#### stopRemainingChunksOnFailure
+
+Stops the run when a chunk exhausts its retries. Off by default (remaining chunks
+still run).
+
+**Signature**
+
+```apex
+ChunkBuilder stopRemainingChunksOnFailure();
+```
+
+#### priority
+
+Sets the chunk job priority.
+
+**Signature**
+
+```apex
+ChunkBuilder priority(Integer priority);
+```
+
+#### delay
+
+Sets a one-time delay in minutes before the first page runs. It does not repeat
+on later pages.
+
+**Signature**
+
+```apex
+ChunkBuilder delay(Integer delay);
+```
+
+#### delayBetweenChunks
+
+Throttles the run by waiting this many minutes before each page after the first.
+Use it to spread DML/callout load or stay under per-hour async limits. Salesforce
+caps enqueue delay at 10 minutes.
+
+**Signature**
+
+```apex
+ChunkBuilder delayBetweenChunks(Integer minutes);
+```
+
+#### retry
+
+Sets the maximum retry attempts per chunk. Must be `0..10`.
+
+**Signature**
+
+```apex
+ChunkBuilder retry(Integer maxRetries);
+```
+
+#### backoff
+
+Sets the retry backoff strategy. See the [Queueable backoff table](/api/queueable#backoff).
+
+**Signature**
+
+```apex
+ChunkBuilder backoff(Backoff backoff);
+```
+
+#### retryOn
+
+Restricts retries to the given exception type(s).
+
+**Signature**
+
+```apex
+ChunkBuilder retryOn(Type exceptionType);
+ChunkBuilder retryOn(List<Type> exceptionTypes);
+```
+
+#### mockId
+
+Sets a mock id for the chunk job. See [AsyncMock](/api/async-mock).
+
+**Signature**
+
+```apex
+ChunkBuilder mockId(String mockId);
+```
+
+#### keepChunkPages
+
+By default a settled page is pruned from the chain once its result is recorded,
+so the run stays flat at any scale (each page still writes its own
+`AsyncResult__c` when result tracking is enabled). Call this to retain settled
+pages in `queueableChainState.jobs` for introspection instead. Off by default;
+not for large runs, since kept pages grow the serialized chain state.
+
+**Signature**
+
+```apex
+ChunkBuilder keepChunkPages();
+```
+
+### Execute
+
+#### enqueue
+
+Enqueues the run and returns the `Async.Result` for the first page. Enqueuing an
+empty source is a safe no-op (nothing is enqueued and `result.salesforceJobId` is
+`null`).
+
+**Signature**
+
+```apex
+Async.Result enqueue();
+```
+
+**Example**
+
+```apex
+Async.Result result = Async.chunk(new AccountRecalcJob(), ChunkSource.of(records))
+	.chunkSize(50)
+	.enqueue();
+```