byrcsc/laravel-data-sync · 1.x

Queues and workers.

How a file becomes a queued batch, how jobs are routed to a connection and queue, what cancellation does, and what a multi-server deployment requires.

The queued path is what sync:run uses by default. It exists so one large file can be processed by several workers, and so a supplier's 2 a.m. delivery does not hold a scheduler process open for twenty minutes.

From file to batch

For each discovered file, sync:run records a run and dispatches one job:

ProcessFileJob for a sync definition
TransferFileJob for a transfer

ProcessFileJob plans the file: it reads the header, validates the schema against the definition's fields, resolves duplicates across the whole file, and divides the rows into chunks of chunkSize(). It then dispatches those chunks as a Bus::batch() of ProcessChunkJobs named data-sync:<definition>:<run ulid>, and stores the batch id on the run.

Each chunk job re-reads the staged file, skips to its offset, maps and writes its rows in one transaction, and increments the run's counters atomically. When every chunk finishes, the batch's completion callback finalizes the run: archives the file, applies the source action, sets the final status, and dispatches SyncCompleted.

A definition with atomic() does not batch. ProcessFileJob processes the whole file itself, in one transaction.

Requirements

A worker. php artisan queue:work — nothing is written until one runs.
The job_batches table. Batches are a framework feature and need it; sync:doctor reports its absence as a failure.
A cache store with atomic locks, used by the discovery lock.
A staging disk both processes can read. See below.

Routing

Jobs run on the default connection and queue unless configured otherwise. Three places can set them, most specific first:

data-sync.overrides.<name>.queue
The definition's queueConnection() and queue() methods
data-sync.queue

// config/data-sync.php
'queue' => [
    'connection' => 'redis',
    'queue' => 'default',
],

'overrides' => [
    'products' => [
        'queue' => ['connection' => 'redis', 'queue' => 'imports'],
    ],
],

// on the definition
public function queueConnection(): ?string
{
    return 'redis';
}

public function queue(): ?string
{
    return 'imports';
}

The override also accepts a bare string, which sets the queue name only:

'overrides' => [
    'products' => ['queue' => 'imports'],
],

Routing applies to the file job, every chunk job, and the batch itself, so one definition's work stays on one queue.

Give large or slow feeds their own queue. A single 500,000-row file becomes hundreds of chunk jobs, and on a shared queue those sit in front of everything else your application dispatches.

Failure and cancellation

What happens	Result
A chunk job throws	Its transaction rolls back; the batch's catch callback fails the run
The batch is cancelled	Remaining chunk jobs return immediately; the run is marked `cancelled`
`maxErrors()` is exceeded	The chunk job cancels the batch and throws; the run fails
The file job throws before batching	The run is failed and the staged file is preserved

A failed or cancelled run preserves the staged file on the failed disk and records its path, so sync:retry can restore and reprocess it even after the source file is gone.

Chunks that already committed stay committed. Reprocessing the file — by retry, or by the next sync:run, since a failed checksum is not blocked — writes them again. That is safe for Upsert and UpdateOnly definitions and worth thinking through for CreateOnly ones.

Queue-level retries behave the same way: a retried chunk job re-reads its rows and rewrites them, and counters are incremented only after the chunk's transaction commits.

Multi-server deployments

Staging is shared state. sync:run stages each discovered file in the dispatching process, and the worker that processes it reads from the staging disk. When the dispatcher and the worker are on different machines, a local staging disk means the worker looks for a file that is not there.

Point data-sync.staging.disk, data-sync.archive.disk, and data-sync.failed.disk at storage every node can reach — S3, a shared NFS mount, anything not local to one box:

'staging' => ['disk' => 's3-internal', 'path' => 'data-sync/staging'],
'archive' => ['disk' => 's3-internal', 'path' => 'data-sync/archive'],
'failed'  => ['disk' => 's3-internal', 'path' => 'data-sync/failed'],

sync:doctor warns whenever the staging disk uses the local driver. On a single-machine deployment that warning is expected and can be ignored — which is worth knowing before you add --strict to a deploy pipeline.

The same applies to sync:retry: it restores from the archive or failed disk into staging, then dispatches a job. Both ends must see the same storage.

Monitoring in flight

php artisan sync:status

For runs with a batch, sync:status reads the batch record and shows real progress — 42% (84/200 jobs) — rather than a row count. See run history and status.

sync:doctor warns about runs that have been running for more than an hour, which is usually a worker that died mid-batch.