JAMES-4231 S3 Object Compaction - #3193
HesandaLiyanage wants to merge 24 commits into
Conversation
|
Hi @HesandaLiyanage ! First of all thank you very much for trying to solve this. However 6k lines changes in one commit is too much for code readability and reviewing properly. Can you cut this work down to smaller step commits? Thank you :) |
|
And test errors are related to this PR I believe, please have a look too : https://ci-builds.apache.org/job/james/job/ApacheJames/job/PR-3193/1/testReport/ |
Document the context, architecture, chunk format, crash safety invariants, and garbage collection strategy for autonomous S3 object compaction.
- Add BlobStoreDAO.readRange(bucket, blobId, offset, endOffset) returning a Mono<RangeByteSlice> with total object size and requested slice. - Define default readRange fallback using readBytes. - Add ChunkMarker utility for discriminating compaction chunk references. - Add contract tests for readRange in ReadSaveBlobStoreDAOContract.
- Implement S3BlobStoreDAO.readRange using AWS S3 HTTP Range requests. - Handle S3 416 RequestedRangeNotSatisfiable when requested offset exceeds object size or range is invalid, falling back or erroring appropriately. - Add contract and MinIO integration tests for S3 range reads.
- Update BloomFilterGCAlgorithm to skip ChunkMarker chunk references during GC sweeps so that compacted chunks are never treated as unreferenced blobs. - Add test coverage in BloomFilterGCAlgorithmContract.
…and compaction tasks - Implement chunk packing (~100MB chunks) with per-slot Zstandard compression and 64KB suffix-indexed footers. - Introduce ChunkId and BlobSlot representations for slot addressing. - Implement ChunkedBlobStoreDAO for transparent single-request HTTP ranged reads and self-healing slot resolution via BlobIdRepairer. - Implement BlobCompactionAlgorithm with windowed candidate payload streaming (DEFAULT_CANDIDATE_BATCH_SIZE = 1000) to bound heap usage during compaction. - Implement BlobCompactionTask, DTOs, and full unit test suite.
…bAdmin API - Implement CassandraBlobReferenceMappingSource, CassandraBlobIdUpdater, and CassandraBlobIdRepairer for updating and self-healing blob references. - Wire BlobCompactionModule in Guice distributed server configuration. - Expose DELETE /blobs?scope=compaction&generation=<gen>&family=<fam> in BlobRoutes to schedule BlobCompactionTask via WebAdmin. - Add integration tests for Cassandra repair/update and WebAdmin compaction routes.
56e7fb9 to
71d5234
Compare
|
Hi @Arsnael, Thank you very much for the review and guidance! 1. CI Test Failures FixedThe failures in
2. Broken Down Into 6 Logical Step CommitsI have restructured the work from a single monolithic commit into 6 focused, incremental commits:
Please let me know if you would like any further adjustments! |
chibenwa
left a comment
chibenwa
left a comment
There was a problem hiding this comment.
Hello,
Thanks a lot for this initial contribution that is going overall in the good direction.
I did do a quick initial review of this work, and have already significant structural comments that I would like to see addressed.
Also I believe we would need to have more integration tests:
Given X, Y Z 200 mail population, compaction yield x, y, z object layout afterward, and we matches integrity constraints (no metadata -> 404). We would need that to be performed on ~5 different layouts, probably on top of memory blob store for decent testing time.
We would also need tests that account for failures updating references and ensure that we remain coherent.
I to not think I saw mailbox/cassandra ObjectStorageNotFoundException upon reading blobs from imapUidTable / messageIdTable - this detail is important for correctness as these table can create entry with past generation blobs (COPY / MOVE)
We are missing /docs documentation for this.
Wouldn't you mind addressing those feedbacks?
| } | ||
| } | ||
|
|
||
| default Mono<RangeByteSlice> readRange(BucketName bucketName, BlobId blobId, long start, long end) { |
There was a problem hiding this comment.
| default Mono<RangeByteSlice> readRange(BucketName bucketName, BlobId blobId, long start, long end) { | |
| default Mono<Blob> readRange(BucketName bucketName, BlobId blobId, long start, long end) { |
This also means chunk reading including:
- blob metadata
- crc validation
is a responsibility of the blob store.
There was a problem hiding this comment.
Done. readRange now returns Publisher<Blob> preserving metadata and payload, allowing ChunkedBlobStoreDAO to propagate blob metadata through the blob store hierarchy.
| public byte[] decompress() throws ObjectStoreIOException { | ||
| if (originalSize < 0 || originalSize > Integer.MAX_VALUE) { | ||
| throw new ObjectStoreIOException("Invalid original size in chunk slot: " + originalSize); | ||
| } | ||
| byte[] decompressed; | ||
| if (originalSize == 0) { | ||
| decompressed = new byte[0]; | ||
| } else { | ||
| try { | ||
| decompressed = Zstd.decompress(compressedContent, (int) originalSize); | ||
| } catch (Exception e) { | ||
| throw new ObjectStoreIOException("Failed to decompress slot content at " + contentStart, e); | ||
| } | ||
| } | ||
|
|
||
| if (decompressed.length != originalSize) { | ||
| throw new ObjectStoreIOException("Decompressed size " + decompressed.length + " does not match expected " + originalSize); | ||
| } | ||
|
|
||
| CRC32C crc = new CRC32C(); | ||
| crc.update(decompressed); | ||
| int computedCrc = (int) crc.getValue(); | ||
| if (computedCrc != crc32c) { | ||
| throw new ObjectStoreIOException(String.format("CRC mismatch for chunk slot at offset %d: expected %d, got %d", | ||
| contentStart, crc32c, computedCrc)); | ||
| } | ||
|
|
||
| return decompressed; | ||
| } |
There was a problem hiding this comment.
We shall not decompress here this is a responsibility of the ZstdBlobStoreDAO
We should instead cary over the metadata for the ZstdBlobStoreDAO to do its job.
There was a problem hiding this comment.
Done. Decompression has been removed from BlobSlot and ChunkedBlobStoreDAO, preserving the original content-encoding and content-original-size metadata in chunk slots so that ZstdBlobStoreDAO handles decompression transparently.
| * reference lookups in future iterations. | ||
| * </p> | ||
| */ | ||
| private Mono<Map<BlobId, Set<String>>> loadReferenceMapping() { |
There was a problem hiding this comment.
This reference mapping is too large to be held in memory.
The proposal was to add a server/data-api interface for it and a Cassandra implementation - I presume.
There was a problem hiding this comment.
Done. Candidate references are now queried on-demand in windowed batches from BlobReferenceMappingSource (implemented by CassandraBlobReferenceMappingSource) rather than loading the entire mapping table into memory.
| public Mono<CompactionResult> compact(CompactionRequest request) { | ||
| return initialCompact(request) | ||
| .flatMap(initialResult -> gcCompact(request).map(initialResult::combine)); | ||
| } |
There was a problem hiding this comment.
I believe we need to have two distinct tasks, one for initial compaction of a generation
And one other for compacting already compacted generation.
There was a problem hiding this comment.
Done. Compaction has been split into two distinct tasks: InitialBlobCompactionTask (for packing standalone blobs of a completed generation into chunks) and GCBlobCompactionTask (for reclaiming space from dead slots in already compacted chunks and merging small chunks). Both tasks have their own dedicated DTOs and WebAdmin scopes.
| return Flux.from(rawStore.listBlobs(request.bucketName())) | ||
| .filter(blobId -> matchesGenerationAndFamily(blobId.asString(), request.generation(), request.family())) |
There was a problem hiding this comment.
We can limit this with a prefix ?
At list offer an option for prefix vs post filtering.
There was a problem hiding this comment.
Done. Added generation prefix pushdown when listing blobs using BlobStoreDAO.listBlobs(bucket, prefix) with fallback to post-filtering when prefix listing is unsupported.
| .filter(blobId -> matchesGenerationAndFamily(blobId.asString(), request.generation(), request.family())) | ||
| .filter(blobId -> !ChunkId.isChunkRef(blobId)) | ||
| .filter(mapping::containsKey) | ||
| .window(DEFAULT_CANDIDATE_BATCH_SIZE) |
There was a problem hiding this comment.
Windowing in terms of count-of-items vs windowing in terms of sum-of-item-size ?
There was a problem hiding this comment.
Done. Windowing is now based on cumulative byte size up to chunkTargetSize rather than pure item count.
| LOGGER.warn("Failed reading candidate blob {}", blobId.asString(), error); | ||
| return Mono.empty(); | ||
| })) | ||
| .filter(candidate -> candidate.payload.length < request.configuration().maxPackableSize()) |
There was a problem hiding this comment.
Imo we need to do this filtering prior windowing
There was a problem hiding this comment.
Done. Candidate blobs exceeding maxPackableSize are now filtered out prior to windowing and kept as standalone objects.
| if (candidates.isEmpty()) { | ||
| return Mono.just(CompactionResult.NONE); | ||
| } |
There was a problem hiding this comment.
Handle also candidate = 1 with no packing while we are at it.
There was a problem hiding this comment.
Done. Batches containing only 1 candidate blob are skipped without packing into chunks.
| ChunkId chunkId = ChunkId.ofChunk(family, request.generation()); | ||
|
|
||
| List<BlobSlotContent> slots = batch.stream() | ||
| .map(candidate -> BlobSlotContent.of(candidate.payload)) |
There was a problem hiding this comment.
We need to account for metadata
There was a problem hiding this comment.
Done. Slot headers and chunk format now serialize and restore blob metadata for each slot.
| .concatMap(i -> { | ||
| CandidateBlob candidate = batch.get(i); | ||
| SlotRange range = writeResult.slotRanges().get(i); | ||
| ChunkId slotRef = ChunkId.slotRef(chunkId, range.offset(), range.limit()); | ||
| Collection<String> messageIds = mapping.getOrDefault(candidate.blobId, Set.of()); | ||
| return blobIdUpdater.replaceReferences(candidate.blobId, slotRef, messageIds) | ||
| .thenReturn(Optional.of(candidate)) | ||
| .onErrorResume(e -> { | ||
| LOGGER.error("Failed to update references for candidate blob {}, skipping deletion of original blob", candidate.blobId.asString(), e); | ||
| return Mono.just(Optional.empty()); | ||
| }); | ||
| }) |
There was a problem hiding this comment.
Extract this in a dedicated method
There was a problem hiding this comment.
Done. Extracted reference replacement and original blob deletion into dedicated method updateReferencesAndDeleteOriginalBlobs.
|
hello, The structure looks better yes thank you @HesandaLiyanage https://ci-builds.apache.org/job/james/job/ApacheJames/job/PR-3193/2/ New test failures still related to this work. It seems some guice bindings are missing:
Requested by: Learn more:
Requested by: Learn more: 2 errors |
- Update BlobStoreDAO.readRange to return Mono<Blob> instead of RangeByteSlice. - Store totalObjectSize in BlobMetadata (TOTAL_OBJECT_SIZE) and provide BlobStoreDAO.totalObjectSize(Blob) utility method. - Update S3BlobStoreDAO and ChunkedBlobStoreDAO to return Mono<Blob> with metadata. - Update contract and unit test suites to use Blob.
…nto distinct tasks
…ound in message metadata
|
All requested changes and feedbacks have been addressed and pushed as individual atomic commits: 1. Structural & Algorithmic Improvements
2. Guice Dependencies in Non-Cassandra Environments
3. Cassandra Mailbox Mappers Fallback
4. Coherence Under Failure Integration Tests
5. Layout Matrix Integration Tests (200 Mails)
6. Documentation
All changes are completed and tested. Thank you both for the detailed review! |
|
Thanks for all the fixes. That's a major changeset. I really woupd support (out of the CI) Property Based Testing for this so that we can develop confidence on it. ( https://www.scalatest.org/user_guide/property_based_testing for instance ) it should be able to generate large set of object within a generation, perform compact and delete in random manner under various profiles (delete heavy or not - configurable and tested accross a wide range of value, possibly randomly set between each compacts) and ensure along several runs that 1- count of object decrease in a predicable fashion and 2- no dead references are created. The review is massive, and I do not want to delegate this to an AI, I'll try to cover a review of this by the end of the week. Cheers, Benoit |
Thanks so much, really appreciate you taking the time on this. Yeah let's do the property based testing. that's way better than me just eyeballing it after a few manual runs. i'll start working around with that on my side in the meantime. looking forward to your feedback by end of week, no rush though given how big this changeset is. |
|
@HesandaLiyanage I see you being a student in your github profile. Would you be interested in a Google Summer of Code with the ASF James project ? If this is meaningful for you we could keep going on the S3 compaction stuff and plan eventually S3 GC / compaction implemented as Apache Spark jobs (for instance). I think the timing is not bad for 2027, though I recognize I do not have the precise schedule in mind. |
Yes, absolutely! I would love to participate in Google Summer of Code with James project. Taking on S3 compaction and exploring large-scale S3 GC/compaction with Apache Spark sounds like a fantastic project, and I'd be honored to work on it with you and the community. |
I'm glad to hear that. This discussion cannot stay solely on github. I propose you to introduce yourself on the DEV mailing list, quote your objectives, link this PR, and potentially say that you would eventually be interested by a GSOC proposed by Benoit to continue the work started on S3 compaction task but make it more scalable with Spark. |
Sounds good! I'll introduce myself on the dev mailing list and keep working on the S3 compaction for now. Thank you so much for the opportunity . |
|
https://ci-builds.apache.org/job/james/job/ApacheJames/job/PR-3193/3/testReport/ Related errors on CassandraBlobIdRepairerIntegrationTest. Please help review and fix @HesandaLiyanage :) |
|
Which S3 should be used now? SeaweedFS? Silo? RustFS? Garage? ofc i dont want get my data corrupted. |
That's a good question. We had a try at triaging alternatives to MinIO earlier this year, as you can see here: https://issues.apache.org/jira/browse/JAMES-4167 It wasn't very conclusive back then unfortunately. But maybe now it's worth taking an other look if you want a nice little extra challenge ;) Specially now that MinIO removed all their public images it seems... I cans ee RustFS for example now released in a v1.0.0 not long ago. Maybe that means they are finally on a more stable and mature version of their product. It would be worth getting an other shot at it IMO |
i tested with James RustFS and for now its bad, about 60-70% compability (used warp to test) also not even close to Silo (MinIO fork 99-100%) and SeaweedFS (which is about 85% combability) perfomance (but yeah much lower ram usage) I tesing RustFS 1.0.0 version, both original and from sources. But you right... there NO real options for now, for standalone single server setup (postgres-app) i think need focus on RustFS combability, for clusters (distribution-app) CephFS i think? Because SeaweedFS dont have full claster support in opensources version, Garage much worse than Seaweed in any way (compability/perfomance, latency) and not working on Windows (but i DO custom version and make it WORK on windows from sourses and even THIS show better than RustFS...) So what you think? |
|
Also why only S3 or PostgreSQL? Why not filesystem like in Stalwart and Dovecot? |
Done!. |
so now not James itself makes compression - now it doings S3 storage? |
No, James itself still does all the compression. S3 is just an object store that receives and stores the bytes James sends it.
|
well i looking forward to move this task to RustFS S3 for example which can compress by myself with zstd/lz4, just as for me the more mail server dont do others jobs the better it doing main job - mail exchanging. |
yeah fair enough, and compression in james isn't forced either way, if the underlying s3 engine you're using already handles it at the storage layer, ceph or rustfs, you can just turn it off on the james side with blobstore.compression.enabled=false and let the backend do its thing. but for plain cloud s3 like aws, wasabi, backblaze b2 etc, none of those do any packing or compression on their own, so having james handle it out of the box still matters a lot there, otherwise your put costs and storage footprint climb pretty fast. |
yeah customizations always wins! so yeah for cloud its sure userful, also only zstd for now? i think in some cases lz4 needed (lower latency). And deduplication methods its like in ZFS? Just Windows Post-processing dedup wins for now. |
Good questions, though i think this is kinda drifting outside the scope of this specific PR, lz4 vs zstd tradeoffs and dedup approaches would probably get a lot more visibility on the dev mailing list or as a github discussion rather than buried in a PR thread. Really appreciate you thinking through this stuff though, i think its better if we keep this thread focused on the compaction review and the CI fixes for now |
We used to have a poorly performing file maildir backend. We deprecated it: bad performance, directory traversal, no maintainer, no known deployment.
If options exist to tell s3 to compress and do the job that's nice. I am not aware TBH and we could add support for this in S3BlobStoreDAO ZstdBlobStoreDAO should work out of the box atop file implem. If this is not the case a separated PR is welcomed.
Bad choice, it won't compress base64 as it doesn't do entropy coding. Most benefit of compression is the base64 overhead on large attachment, which is capped at ~33%.
If it is standard (ie amazon s3 supports this) then it is welcomed in S3BlobStoreDAO
Yes precisely. Also bear in mind that one can implement out-of-band-compression, and solely compress old 3+ month old mail as a tiering strategy... I do not know if RustFS for instance is able to automate that with a S3 policy.
Event worse if one operate a on prem S3 compatible system like Ceph Rados Gateway.
Lz4 is not fit cf earlier entropy coding point. Zstd had decent java port. More niche algorithm would need to have good perf java implems. I confess I do not have in mind viable java alternative, but with hashing I did bench SHA-256 vs BLAKE3 and fornow JVM implenm of BLAKE3 is worse than SHA-256. I'd be careful with exotic implem and have their viability backed with a JMH micro benchmark.
You are free to drop james dedup and experiment with FS level dedup. My take on it is that FS do not have all the infos James have and would be doing more work to achieve partial results. I'm expecting it not to deduplicate as well as James can do but I do not have solid comparison to answer you and the question indeed is interesting.
Completly, but this is a nice discussion! Cheers, Benoit |
I confirm this is the right way to go! |
While I 100% subscribe to the point as a James integrator, |
|
thanks for laying all that out, @chibenwa , appreciated. for file/postgres, since ZstdBlobStoreDAO is just a decorator, getting it properly wired and tested on top of FileBlobStoreDAO and PostgresBlobStoreDAO seems like its own clean follow up PR too. I can work on that |
4 participants
JIRA: https://issues.apache.org/jira/browse/JAMES-4231
Context & Problem
High email ingest volumes in Apache James deployments utilizing S3/MinIO object storage result in hundreds of millions of small S3 objects (average body/header size ~20–50KB). Operating at this granularity introduces high S3 API request fees (PUT/GET/LIST), elevated bucket listing latency during storage maintenance and garbage collection, and reduced overall object storage throughput.
Solution Overview
This PR implements autonomous S3 object compaction for Apache James, packing historical standalone blobs into large, immutable chunk files (~100MB target size) with slot-level virtual addressing and transparent single-request HTTP ranged read support.
Key Components & Architecture
Chunk Format (
ChunkFormat):[Header (16B)][Slot 0]...[Slot N-1][Footer][Footer Size (4B)][End Magic 0x454F4643]content-encoding=zstd\n), allowing single-slot ranged reads without decompressing or buffering the full 100MB chunk.readRange(..., -65536, -1)) fetches chunk slot metadata insrc/adr/0076-s3-object-compaction.md.Virtual Slot Addressing & Ranged Reads (
ChunkedBlobStoreDAO&ChunkId):<family>_<generation>_chunk<hash>~<offset>~<limit>.ChunkedBlobStoreDAOtransparently intercepts slot references and issues HTTP byte-range requests directly against the underlying storage connector (readRange(bucket, chunkId, offset, offset + limit - 1)).ChunkMarkerregex (^\d+_\d+_chunk[A-Za-z0-9_-]{16,}(~\d+~\d+)?$) to ensure bloom-filter GC never misclassifies chunks as unreferenced garbage.Compaction Pipeline & Crash Safety (
BlobCompactionAlgorithm):initialCompact):DEFAULT_CANDIDATE_BATCH_SIZE = 1000) so candidate byte arrays are packed and freed window-by-window without buffering the generation's payloads in heap.messageV3,messageIdToImapUid,messageIdTable).gcCompact):CassandraBlobIdRepairer):blobReferenceMappingand restores Cassandra references. Also heals any transient inconsistency window across Cassandra tables.Configuration Guards & Scope:
DELETE /blobs?scope=compaction&generation=<gen>&family=<fam>.Memory Characteristics & Operational Ceiling:
Verification & Tests
server/blob/blob-compaction: 47 tests passed (100%), including allocation-bound proof test and task serialization.server/blob/blob-storage-strategy: 78 tests passed (100%).mailbox/cassandra(CassandraBlobId*IntegrationTest): 3/3 passed with real Cassandra 5.0.9 testcontainers.server/blob/blob-s3: Ranged read and MinIO S3 end-to-end compaction integration tests passed.server/container/guice/distributed: 43 tests passed (100%).