Custom Storage Backends — Quarkus TUS

The store owns bytes; the framework owns the protocol

A store persists the UploadInfo records it is handed, moves bytes into place, and answers simple questions about what it holds. It never validates a TUS rule, never computes a checksum, never fires an event and never builds a URL — the extension does all of that before and after calling into the store. You can write a store knowing only your storage system, without having read the TUS specification.

Bytes are streamed. A PATCH body flows from the socket, through the extension, into your store as a backpressured Multi<Buffer>; it is never materialised in memory or on disk on the way. That is what lets an S3 store hand each chunk to UploadPart as it arrives.

The UploadStore Interface

org.sitenetsoft.quarkus.tus.runtime.spi.UploadStore:

public interface UploadStore {

    // Records — the framework builds them, the store persists them
    Uni<Optional<UploadInfo>> findUploadInfo(String id);
    Uni<String> createUpload(UploadInfo info);            // returns an id, never a URL
    Uni<Void> updateUploadInfo(String id, UploadInfo info);

    // Bytes — a staged write
    Uni<Long> stageChunk(String id, long offset, Multi<Buffer> data, long expectedLength);
    Uni<Void> commitChunk(String id, long offset, long bytesStaged);
    Uni<Void> abortChunk(String id, long offset);

    // Concatenation — join sources into a final record the framework already created
    Uni<Void> concatenate(String finalId, List<String> sourceIds);

    // Lifecycle and locking
    Uni<Boolean> discardUpload(String id);
    Uni<Boolean> acquireLock(String id);
    Uni<Void> releaseLock(String id);

    // Maintenance
    Uni<List<String>> cleanupExpiredUploads();
    default Uni<Void> cleanupStaleLocks() { return Uni.createFrom().voidItem(); }
    default Uni<List<String>> cleanupStaleUploads(long staleHours) { return Uni.createFrom().item(List.of()); }
    default Uni<Integer> cleanupOrphanFiles() { return Uni.createFrom().item(0); }
}

Why writes are staged

The checksum extension requires that a chunk with a bad Upload-Checksum be answered with 460 and the upload’s offset unchanged — the bytes must not count. When bytes stream, the digest is only known once the last byte has already reached storage. So a write happens in two steps:

stageChunk streams the bytes into place at offset but must not advance the upload’s offset. findUploadInfo between stage and commit shows the old offset.
commitChunk makes them part of the upload (offset becomes offset + bytesStaged, lastActivity is stamped, the record is persisted); or abortChunk discards them and leaves the upload exactly as it was.

The framework holds the upload’s lock around the whole sequence, computes the digest as the bytes pass through, and decides commit or abort. Your store never sees a checksum, and it gets correct checksum behaviour for every algorithm for free.

stageChunk commitChunk abortChunk

	`stageChunk`	`commitChunk`	`abortChunk`
Local file	write at `offset`	set offset, persist record	truncate the file back to `offset`
S3 multipart	`UploadPart` (needs `expectedLength`)	record the part’s ETag	drop the part
Another TUS server, buffered	collect the chunk in memory	`PATCH` it downstream	drop the buffer
Another TUS server, pass-through	`PATCH` the bytes downstream as they arrive	record the new remote offset	cannot un-send: `DELETE` the remote upload and re-create it (see below)

Local file

write at offset

set offset, persist record

truncate the file back to offset

S3 multipart

UploadPart (needs expectedLength)

record the part’s ETag

drop the part

Another TUS server, buffered

collect the chunk in memory

PATCH it downstream

drop the buffer

Another TUS server, pass-through

PATCH the bytes downstream as they arrive

record the new remote offset

cannot un-send: DELETE the remote upload and re-create it (see below)

expectedLength is the declared chunk length, or -1 when there is none (chunked transfer encoding, or a length-less HTTP/2 body). The framework never stages a chunk it knows to be empty; a length-less body that turns out to be empty is staged as zero bytes and then aborted, not committed.

Contract in brief

stageChunk at an offset that is not the upload’s current one must fail with OffsetMismatchException. The framework has already validated it under the lock; a mismatch here means a request raced past validation, and writing there would overwrite acknowledged bytes.
Anything asked about an unknown id: findUploadInfo returns empty, updateUploadInfo is a no-op, stageChunk/concatenate fail with UploadNotFoundException, discardUpload returns false.
abortChunk is idempotent and safe to call when nothing was staged or when staging itself failed.
concatenate(finalId, sourceIds) fills a final upload the framework already created with isFinalConcat=true and partialIds; on success set its offset to its length, clear isFinalConcat and partialIds, persist. Leave the sources alone — the framework discards them.
discardUpload just deletes: the framework holds the upload’s lock whenever it calls it (DELETE takes the lock; a finished concatenation discards its partials under the locks it already holds), so the store neither takes nor checks the lock there. Your own cleanup (cleanupExpiredUploads and the hooks) must take the lock before deleting, so that it never removes an upload underneath an in-flight write.
A failure of stageChunk — unknown id, stale offset, or a failure that came down the body stream (client gone, limit crossed) — is a failure of the returned Uni, propagated as-is; never a synchronous throw, never wrapped. The framework decides the response from the failure’s type. After the framework’s abortChunk, the upload is exactly as it was.
The lock is held for the whole of stageChunk, i.e. for as long as the client takes to send the chunk. A store that expires abandoned locks must treat a lock as live while bytes are flowing (the bundled store refreshes it on every buffer) and give it a timeout longer than any pause a healthy client may make mid-chunk — see quarkus.tus.lock-timeout-seconds.
Every method is subscribed to on a Vert.x event loop: do not block in any of them. Use your backend’s asynchronous client, or push blocking work onto a worker (vertx.executeBlocking(…), uni.runSubscriptionOn(Infrastructure.getDefaultWorkerPool())). Records that live in memory just return Uni.createFrom().item(…); records that live in a database or another service can be a round trip, so a store needs no local index. (BufferingUploadStore runs your appendBytes on a worker for exactly this reason.)
createUpload returns an id. It is the last segment of the Location the extension builds; it must not contain /.

Every one of these rules is an assertion in the contract test below.

Implementing a Custom Backend

Provide a CDI bean implementing UploadStore. LocalFileUploadStore is a @DefaultBean, so a plain @ApplicationScoped implementation replaces it by merely existing — no @Alternative or @Priority is needed for a production replacement, though both still work as before. For test-only stores that must not replace the bundled one in every test, annotate the bean @Alternative (without @Priority) and enable it per test via QuarkusTestProfile.getEnabledAlternatives() (see Testing).

The easy way: `BufferingUploadStore`

If your backend appends whole byte arrays and you do not mind holding one chunk per in-flight upload in memory, extend org.sitenetsoft.quarkus.tus.runtime.spi.BufferingUploadStore. It implements the staged write for you — stageChunk collects the stream, commitChunk calls your appendBytes and advances the record, abortChunk drops the buffer — so you implement only the record methods, appendBytes, concatenate, discardUpload, the lock pair and cleanupExpiredUploads:

@ApplicationScoped
public class MyStore extends BufferingUploadStore {

    private final Map<String, UploadInfo> records = new ConcurrentHashMap<>();
    private final Set<String> locks = ConcurrentHashMap.newKeySet();

    @Override public Uni<Optional<UploadInfo>> findUploadInfo(String id) { return Uni.createFrom().item(Optional.ofNullable(records.get(id))); }
    @Override public Uni<String> createUpload(UploadInfo info) { String id = UUID.randomUUID().toString(); records.put(id, info); return Uni.createFrom().item(id); }
    @Override public Uni<Void> updateUploadInfo(String id, UploadInfo info) { records.computeIfPresent(id, (k, v) -> info); return Uni.createFrom().voidItem(); }

    @Override protected void appendBytes(String id, long offset, byte[] data) { /* append to your storage; runs on a worker */ }
    @Override public Uni<Void> concatenate(String finalId, List<String> sourceIds) { /* join, then mark finalId complete */ }

    @Override public Uni<Boolean> discardUpload(String id) { /* remove bytes + record; the caller holds the lock */ }
    @Override public Uni<Boolean> acquireLock(String id) { return Uni.createFrom().item(locks.add(id)); }
    @Override public Uni<Void> releaseLock(String id) { locks.remove(id); return Uni.createFrom().voidItem(); }
    @Override public Uni<List<String>> cleanupExpiredUploads() { /* remove records whose expiresAt is past */ }
}

The in-memory store used by the extension’s own test suite is exactly this, in about a hundred lines, and injects nothing.

The streaming way: an S3 sketch

A complete, working version of this sketch — S3UploadStore in the extension’s integration-tests module — runs the SPI contract test and a 400 MB upload against MinIO (set TUS_S3_ENDPOINT). It cuts the body into 5 MB multipart parts as the bytes arrive and keeps only the sub-part tail between commits, so a 400 MB chunk goes through a 512 MB JVM holding one part at a time. It is a sample to copy, not a published artifact: the extension ships no vendor SDK.

@ApplicationScoped
public class S3UploadStore implements UploadStore {

    @Inject S3AsyncClient s3;   // your dependency, not the extension's

    @Override
    public Uni<Long> stageChunk(String id, long offset, Multi<Buffer> data, long expectedLength) {
        UploadInfo info = findUploadInfo(id).orElse(null);
        if (info == null) {   // failures are failures of the Uni, never synchronous throws
            return Uni.createFrom().failure(new UploadNotFoundException(id));
        }
        if (offset != info.getOffset()) {
            return Uni.createFrom().failure(new OffsetMismatchException("stale offset", info.getOffset()));
        }
        // S3 needs a content length per part; expectedLength is -1 for chunked bodies, in
        // which case a real store would fall back to buffering that one chunk.
        AsyncRequestBody body = AsyncRequestBody.fromPublisher(
                AdaptersToFlow.publisher(data.map(b -> b.getByteBuf().nioBuffer())));
        return Uni.createFrom().completionStage(s3.uploadPart(uploadPartRequest(id, offset, expectedLength), body))
                .invoke(resp -> rememberPendingEtag(id, offset, resp.eTag()))
                .replaceWith(expectedLength);
    }

    @Override
    public Uni<Void> commitChunk(String id, long offset, long bytesStaged) {
        // Record the ETag against the part number, advance the offset, persist the record.
    }

    @Override
    public Uni<Void> abortChunk(String id, long offset) {
        // Forget the pending ETag; the orphaned part is dropped when the multipart upload completes or aborts.
    }

    @Override
    public Uni<Void> concatenate(String finalId, List<String> sourceIds) {
        // UploadPartCopy each source into finalId's multipart upload, then complete it.
    }

    // records, discard, locks, cleanup ...
}

For a clustered deployment, back acquireLock/releaseLock with something shared (a database row, Redis). The lock is what serialises concurrent PATCHes to one upload; the bundled store’s is per process.

Relaying to another TUS server

A store can forward what it receives to a second TUS server — a downstream service, a different storage tier — using the client extension. The extension ships no such store on purpose; it ships the primitives, and this section is about the one place where the two protocols do not line up.

A TUS server’s offset only moves forward. Once bytes have been PATCH`ed downstream they are part of that upload, and there is no request that takes them back. `abortChunk, however, must leave the upload exactly as it was: the framework calls it whenever a chunk fails its Upload-Checksum, and also when the client disconnects mid-chunk. So a store that forwards bytes while they stream has nothing cheap to do on abort. There are two honest designs:

Buffered: Extend BufferingUploadStore and forward the whole chunk from appendBytes. Staging is the buffer; commit is the downstream PATCH; abort drops the buffer and the downstream server never saw the bytes. Simple and exact, at the cost of holding one chunk per in-flight upload in memory — bound it with quarkus.tus.max-chunk-size.
Pass-through: Stream into the downstream PATCH from stageChunk (TusProtocolClient.patch takes the same Multi<Buffer>), record the offset the downstream server returns in commitChunk, and in abortChunk terminate() the downstream upload, create() a fresh one and replay any already-committed bytes into it — which means keeping a copy of them, or accepting that an abort restarts the upload from zero on the client’s next HEAD. Nothing is materialised on the relay while things go well; the price is paid on abort. Note that a downstream server’s own checksum check does not help here: the framework has already streamed the bytes before it knows the digest, and it is the upstream client’s checksum the framework verifies.

Either way, findUploadInfo is yours to answer. Keep a local record and reconcile it against TusProtocolClient.offset() when they disagree, rather than round-tripping to the downstream server on every HEAD; the lock rules above still apply, so a reconciliation that moves the offset must happen under the upload’s lock.

If neither design suits, do not advertise checksum: a store that cannot honour abortChunk cannot honour a 460 either. See the note on extensions for what each one asks of the store.

Verifying your store: the contract test

The org.sitenetsoft:quarkus-tus-tck artifact contains AbstractUploadStoreContractTest, an abstract JUnit 5 class that checks a store against every rule above. Add it to your test classpath and extend it:

testImplementation 'org.sitenetsoft:quarkus-tus-tck:1.0.0'

@QuarkusTest
class MyStoreContractTest extends AbstractUploadStoreContractTest {

    @Inject
    UploadStore store;

    @Override
    protected UploadStore store() {
        return store;
    }

    // Optional: lets the content assertions run. Without it only offsets are checked,
    // because the SPI has no read API.
    @Override
    protected Optional<byte[]> readBytes(String id) {
        return Optional.ofNullable(myStorage.get(id));
    }
}

Twenty-one assertions run: record round-trips, stage-does-not-advance, commit-advances-exactly, abort-leaves-nothing, stale-offset rejection, unknown-id behaviour, multi-buffer, zero-length and failing-stream stages, concatenation, discard under the caller’s lock, lock exclusivity and expiry cleanup. Both bundled stores pass it, and a store that passes it will behave correctly under the extension.

Default Local File Store

The built-in LocalFileUploadStore stores upload data as files on disk:

Each upload is a file named {uploadId} in the configured directory, with a {uploadId}.meta JSON sidecar holding its record
Chunks are written with Vert.x asynchronous file I/O; abortChunk truncates the file back to the staged offset
Concatenation pipes the source files into the final one without blocking the event loop
Locks are in-memory (single-node only)

Configure the storage directory:

quarkus.tus.store.local.upload-dir=/var/data/uploads

The directory is created automatically at startup if it does not exist.

Note	The default local store is designed for single-node deployments. For clustered environments, implement a custom `UploadStore` with distributed locking (e.g., database-backed or Redis-backed).