Skip to content

Commit 157afb8

Browse files
frristclaudealanshaw
authored
feat(s3): multipart checksums (FIL-620) (#68)
* feat(s3): multipart checksums (FIL-620) Per-part checksums at UploadPart, composite / full-object / default final checksums at Complete, and the ListParts echo, mirroring the upstream posix backend over versitygw's utils (composite checksum-of-checksums reader, CRC combine) and s3err constructors: - multipart_parts gains a checksum column (00009): the session algorithm's base64 value, or the internal full-object CRC64NVME of an undeclared session (never echoed; derives Complete's default final checksum). - UploadPart negotiates the part algorithm against the session declaration (mismatch / composite-without-checksum rejections), computes + validates through a HashReader stack over the ingest pass, persists, and echoes. - Complete verifies the request's checksum type and every part entry (multiple / malformed / missing / wrong-algo / wrong-value), folds the stored part checksums into the final value (COMPOSITE "-N" suffix or FULL_OBJECT CRC combine), verifies a client-supplied final checksum, and persists algorithm/value/type on the manifest; the idempotent re-Complete returns the same checksum fields. - ObjectManifest gains ChecksumType ("cy"); checksumFields stops hardcoding FULL_OBJECT so GET/HEAD/ListObjectVersions echo the stored type (empty = legacy full-object). - ListParts echoes per-part checksums + the session algorithm/type, or the literal "null" pair for an undeclared session. - itest: the 19 checksum rows promote from the UploadPart / ListParts / CompleteMultipartUpload XFail tables to pass; racey_data_integrity stays xfail (load-sensitive concurrency, not checksums). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(s3): CopyObject checksum propagation and replacement A copy without a checksum request carries the source's checksum triple (algorithm/value/type) to the destination; a request naming a different x-amz-checksum-algorithm replaces it, streaming the shared body through the new algorithm once for a full-object value. The CopyObjectResult now reports the destination checksum fields. Promotes the four checksum rows of the CopyObject XFail table. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(s3): release shipped segment blobs on DeleteBucket A shipped catalog segment registers TWO blobs in the bucket's space: the sealed CAR and its sharded-dag-index (both blob/added by SubmitShard). Hilt refuses to delete a space that still holds registrations, so any bucket that lived past the catalog seal age with a successful ship could never be deleted: DeleteBucket returned BucketNotEmpty with no objects left. Surfaced by promoting CompleteMultipartUpload/should_verify_final_composite_ checksum — the first conformance case long enough (~11s of commits) to ship a segment before its teardown. The index blob's digest was recorded nowhere, so ship now persists it: SubmitShard returns the index digest, the flush func hands it to MarkSegmentShipped, and ingot.segments gains an index_digest column (00010). DeleteBucket quiesces the bucket's log first — joining any in-flight ship, so a segment can't register its blobs after the release pass has read the rows (on faster hosts the teardown lands exactly inside the ship window) — then releases every sealed segment's CAR and every shipped segment's index from the space before the hilt delete. Releases are idempotent (removing an unregistered blob is a no-op), so sealed-but-unshipped CARs are released defensively and retries are safe. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Update bucket/manifest.go comment Code comment on relevant code section Co-authored-by: ash <alan138@gmail.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: ash <alan138@gmail.com>
1 parent a607451 commit 157afb8

30 files changed

Lines changed: 1393 additions & 149 deletions

‎blockstore/staging_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,7 @@ func (f *fakeMeta) MarkSegmentSealed(_ context.Context, plane blockstore.Plane,
4242
f.roots = append(f.roots, opRoots...)
4343
return nil
4444
}
45-
func (f *fakeMeta) MarkSegmentShipped(_ context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, _ []blockstore.OpRoot) error {
45+
func (f *fakeMeta) MarkSegmentShipped(_ context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, _ []byte, _ []blockstore.OpRoot) error {
4646
if r, ok := f.rows[seq]; ok {
4747
r.ShippedAt = shippedAt
4848
}
@@ -86,7 +86,7 @@ func (f *fakeMeta) RehydrateSegment(_ context.Context, m logstore.SegmentMeta) e
8686

8787
// nopFlush is the per-plane ship callback for these tests: the store
8888
// owns the ship-state transition, so the closure is a no-op.
89-
func nopFlush(_ context.Context, _ *logstore.Segment) error { return nil }
89+
func nopFlush(_ context.Context, _ *logstore.Segment) ([]byte, error) { return nil, nil }
9090

9191
// noopBase satisfies blockstore.BlockReader but always returns
9292
// errUnknownBase so we can detect when a GetBlock falls through

‎bucket/cbor_gen.go‎

Lines changed: 35 additions & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎bucket/manifest.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,11 @@ type ObjectManifest struct {
3737
// checksum was requested. Independent of the internal sha256 content address.
3838
ChecksumAlgorithm string `cborgen:"ca"`
3939
Checksum string `cborgen:"ck"`
40+
// ChecksumType is the S3 checksum type of Checksum: "FULL_OBJECT" (computed
41+
// over the whole body) or "COMPOSITE" (a multipart checksum-of-checksums
42+
// with a "-N" part-count suffix). Empty in blocks written before the type
43+
// was recorded, which are all full-object.
44+
ChecksumType string `cborgen:"cy"`
4045

4146
// HTTP/S3 system headers carried through PUT and replayed on
4247
// HEAD/GET. Empty strings are omitted from responses.

‎inmem/segments_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@ func TestMarkSegmentShipped_GuardsForgeRootOnRoot(t *testing.T) {
4444

4545
// Ship with the stale op-root LAST: unconditionally (the old behavior) it
4646
// would win as the last write; the guard must skip it and keep `committed`.
47-
if err := m.MarkSegmentShipped(ctx, blockstore.PlaneCatalog, seq, 100, []blockstore.OpRoot{
47+
if err := m.MarkSegmentShipped(ctx, blockstore.PlaneCatalog, seq, 100, nil, []blockstore.OpRoot{
4848
{Bucket: "bk", Root: committed},
4949
{Bucket: "bk", Root: stale},
5050
}); err != nil {
@@ -65,7 +65,7 @@ func TestMarkSegmentShipped_GuardsForgeRootOnRoot(t *testing.T) {
6565
_ = m2.CASRoot(ctx, "bk2", cid.Undef, committed)
6666
seq2, _ := m2.NextSegmentSeq(ctx)
6767
_ = m2.InsertSegmentOpen(ctx, blockstore.PlaneCatalog, seq2, "bk2")
68-
if err := m2.MarkSegmentShipped(ctx, blockstore.PlaneCatalog, seq2, 100, []blockstore.OpRoot{
68+
if err := m2.MarkSegmentShipped(ctx, blockstore.PlaneCatalog, seq2, 100, nil, []blockstore.OpRoot{
6969
{Bucket: "bk2", Root: stale},
7070
}); err != nil {
7171
t.Fatalf("MarkSegmentShipped(stale only): %v", err)

‎inmem/store.go‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import (
1515
"fmt"
1616
"io"
1717
"net/url"
18+
"slices"
1819
"sort"
1920
"strconv"
2021
"strings"
@@ -287,11 +288,12 @@ func (m *MemStore) MarkSegmentSealed(_ context.Context, plane blockstore.Plane,
287288
return nil
288289
}
289290

290-
func (m *MemStore) MarkSegmentShipped(_ context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, opRoots []blockstore.OpRoot) error {
291+
func (m *MemStore) MarkSegmentShipped(_ context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, indexDigest []byte, opRoots []blockstore.OpRoot) error {
291292
m.mu.Lock()
292293
defer m.mu.Unlock()
293294
if r, ok := m.segments[seq]; ok {
294295
r.ShippedAt = shippedAt
296+
r.IndexDigest = slices.Clone(indexDigest)
295297
}
296298
if plane == blockstore.PlaneCatalog {
297299
for _, opr := range opRoots {
@@ -375,8 +377,8 @@ func (NopBaseReader) OpenBlob(_ context.Context, _ did.DID, _ multihash.Multihas
375377
// network, so the spool's local copy serves all reads.
376378
type NopUploader struct{}
377379

378-
func (NopUploader) SubmitShard(_ context.Context, _ blockstore.Plane, _ did.DID, _ uploader.CARShard) (uploader.BlobLocation, error) {
379-
return uploader.BlobLocation{}, nil
380+
func (NopUploader) SubmitShard(_ context.Context, _ blockstore.Plane, _ did.DID, _ uploader.CARShard) (uploader.BlobLocation, multihash.Multihash, error) {
381+
return uploader.BlobLocation{}, nil, nil
380382
}
381383

382384
// UploadBlob accepts immediately, even with WithConclude(false) — there is

‎itest/versity_multipart_test.go‎

Lines changed: 21 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,9 @@ import (
88

99
// Multipart groups of the S3 conformance partition, partitioned empirically
1010
// against the forge-mode stack (see the curation note in README.md). The
11-
// remaining xfail surface: part-level checksums (FIL-620), tagging/object-lock
12-
// /ACL on create (FIL-534/FIL-525), and the UploadPartCopy group (FIL-586).
11+
// remaining xfail surface: tagging/object-lock/ACL on create
12+
// (FIL-534/FIL-525), the UploadPartCopy group (FIL-586), and one
13+
// load-sensitive concurrency case.
1314

1415
var createMultipartPass = []forgeCase{
1516
{name: "non_existing_bucket", fn: integration.CreateMultipartUpload_non_existing_bucket},
@@ -45,18 +46,17 @@ var uploadPartPass = []forgeCase{
4546
{name: "multiple_checksum_headers", fn: integration.UploadPart_multiple_checksum_headers},
4647
{name: "invalid_checksum_header", fn: integration.UploadPart_invalid_checksum_header},
4748
{name: "checksum_header_and_algo_mismatch", fn: integration.UploadPart_checksum_header_and_algo_mismatch},
48-
{name: "success", fn: integration.UploadPart_success},
49-
}
50-
51-
var uploadPartXFail = []forgeCase{
5249
{name: "checksum_algorithm_mistmatch_on_initialization", fn: integration.UploadPart_checksum_algorithm_mistmatch_on_initialization},
5350
{name: "checksum_algorithm_mistmatch_on_initialization_with_value", fn: integration.UploadPart_checksum_algorithm_mistmatch_on_initialization_with_value},
5451
{name: "incorrect_checksums", fn: integration.UploadPart_incorrect_checksums},
5552
{name: "no_checksum_with_full_object_checksum_type", fn: integration.UploadPart_no_checksum_with_full_object_checksum_type},
5653
{name: "no_checksum_with_composite_checksum_type", fn: integration.UploadPart_no_checksum_with_composite_checksum_type},
5754
{name: "with_checksums_success", fn: integration.UploadPart_with_checksums_success},
55+
{name: "success", fn: integration.UploadPart_success},
5856
}
5957

58+
var uploadPartXFail = []forgeCase{}
59+
6060
var uploadPartCopyPass = []forgeCase{
6161
{name: "non_existing_bucket", fn: integration.UploadPartCopy_non_existing_bucket},
6262
{name: "invalid_part_number", fn: integration.UploadPartCopy_invalid_part_number},
@@ -90,13 +90,11 @@ var listPartsPass = []forgeCase{
9090
{name: "truncated", fn: integration.ListParts_truncated},
9191
{name: "success", fn: integration.ListParts_success},
9292
{name: "with_checksums", fn: integration.ListParts_with_checksums},
93-
}
94-
95-
// Explicit null-checksum-type echo is FIL-620.
96-
var listPartsXFail = []forgeCase{
9793
{name: "null_checksums", fn: integration.ListParts_null_checksums},
9894
}
9995

96+
var listPartsXFail = []forgeCase{}
97+
10098
var listMultipartUploadsPass = []forgeCase{
10199
{name: "non_existing_bucket", fn: integration.ListMultipartUploads_non_existing_bucket},
102100
{name: "empty_result", fn: integration.ListMultipartUploads_empty_result},
@@ -142,6 +140,18 @@ var completeMultipartPass = []forgeCase{
142140
{name: "multiple_final_checksums", fn: integration.CompleteMultipartUpload_multiple_final_checksums},
143141
{name: "invalid_final_checksums", fn: integration.CompleteMultipartUpload_invalid_final_checksums},
144142
{name: "invalid_final_composite_checksum", fn: integration.CompleteMultipartUpload_invalid_final_composite_checksum},
143+
{name: "invalid_checksum_part", fn: integration.CompleteMultipartUpload_invalid_checksum_part},
144+
{name: "multiple_checksum_part", fn: integration.CompleteMultipartUpload_multiple_checksum_part},
145+
{name: "incorrect_checksum_part", fn: integration.CompleteMultipartUpload_incorrect_checksum_part},
146+
{name: "different_checksum_part", fn: integration.CompleteMultipartUpload_different_checksum_part},
147+
{name: "missing_part_checksum", fn: integration.CompleteMultipartUpload_missing_part_checksum},
148+
{name: "incorrect_final_checksums", fn: integration.CompleteMultipartUpload_incorrect_final_checksums},
149+
{name: "should_calculate_the_final_checksum_full_object", fn: integration.CompleteMultipartUpload_should_calculate_the_final_checksum_full_object},
150+
{name: "should_verify_the_final_checksum", fn: integration.CompleteMultipartUpload_should_verify_the_final_checksum},
151+
{name: "should_verify_final_composite_checksum", fn: integration.CompleteMultipartUpload_should_verify_final_composite_checksum},
152+
{name: "checksum_type_mismatch", fn: integration.CompleteMultipartUpload_checksum_type_mismatch},
153+
{name: "should_ignore_the_final_checksum", fn: integration.CompleteMultipartUpload_should_ignore_the_final_checksum},
154+
{name: "should_succeed_without_final_checksum_type", fn: integration.CompleteMultipartUpload_should_succeed_without_final_checksum_type},
145155
{name: "with_metadata", fn: integration.CompleteMultipartUpload_with_metadata},
146156
{name: "success", fn: integration.CompleteMultipartUpload_success},
147157
{name: "already_completed", fn: integration.CompleteMultipartUpload_already_completed},
@@ -158,20 +168,7 @@ var completeMultipartPass = []forgeCase{
158168
}},
159169
}
160170

161-
// Part-level / composite-checksum verification is FIL-620;
162-
// racey_data_integrity additionally leans on atomic concurrent overwrites.
171+
// racey_data_integrity leans on atomic concurrent overwrites under load.
163172
var completeMultipartXFail = []forgeCase{
164-
{name: "invalid_checksum_part", fn: integration.CompleteMultipartUpload_invalid_checksum_part},
165-
{name: "multiple_checksum_part", fn: integration.CompleteMultipartUpload_multiple_checksum_part},
166-
{name: "incorrect_checksum_part", fn: integration.CompleteMultipartUpload_incorrect_checksum_part},
167-
{name: "different_checksum_part", fn: integration.CompleteMultipartUpload_different_checksum_part},
168-
{name: "missing_part_checksum", fn: integration.CompleteMultipartUpload_missing_part_checksum},
169-
{name: "incorrect_final_checksums", fn: integration.CompleteMultipartUpload_incorrect_final_checksums},
170-
{name: "should_calculate_the_final_checksum_full_object", fn: integration.CompleteMultipartUpload_should_calculate_the_final_checksum_full_object},
171-
{name: "should_verify_the_final_checksum", fn: integration.CompleteMultipartUpload_should_verify_the_final_checksum},
172-
{name: "should_verify_final_composite_checksum", fn: integration.CompleteMultipartUpload_should_verify_final_composite_checksum},
173-
{name: "checksum_type_mismatch", fn: integration.CompleteMultipartUpload_checksum_type_mismatch},
174-
{name: "should_ignore_the_final_checksum", fn: integration.CompleteMultipartUpload_should_ignore_the_final_checksum},
175-
{name: "should_succeed_without_final_checksum_type", fn: integration.CompleteMultipartUpload_should_succeed_without_final_checksum_type},
176173
{name: "racey_data_integrity", fn: integration.CompleteMultipartUpload_racey_data_integrity},
177174
}

‎itest/versity_object_test.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -173,10 +173,14 @@ var copyObjectPass = []forgeCase{
173173
{name: "invalid_legal_hold", fn: integration.CopyObject_invalid_legal_hold},
174174
{name: "invalid_object_lock_mode", fn: integration.CopyObject_invalid_object_lock_mode},
175175
{name: "invalid_website_redirect_location", fn: integration.CopyObject_invalid_website_redirect_location},
176+
{name: "create_checksum_on_copy", fn: integration.CopyObject_create_checksum_on_copy},
177+
{name: "should_copy_the_existing_checksum", fn: integration.CopyObject_should_copy_the_existing_checksum},
178+
{name: "should_replace_the_existing_checksum", fn: integration.CopyObject_should_replace_the_existing_checksum},
179+
{name: "to_itself_by_replacing_the_checksum", fn: integration.CopyObject_to_itself_by_replacing_the_checksum},
176180
}
177181

178182
// Observed failing against the forge stack: multi-account semantics, tagging,
179-
// object-lock, and checksum-on-copy are unimplemented surface.
183+
// and object-lock are unimplemented surface.
180184
var copyObjectXFail = []forgeCase{
181185
{name: "not_owned_source_bucket", fn: integration.CopyObject_not_owned_source_bucket},
182186
{name: "should_replace_tagging", fn: integration.CopyObject_should_replace_tagging},
@@ -185,10 +189,6 @@ var copyObjectXFail = []forgeCase{
185189
{name: "with_legal_hold", fn: integration.CopyObject_with_legal_hold},
186190
{name: "with_retention_lock", fn: integration.CopyObject_with_retention_lock},
187191
{name: "object_acl_not_supported", fn: integration.CopyObject_object_acl_not_supported},
188-
{name: "create_checksum_on_copy", fn: integration.CopyObject_create_checksum_on_copy},
189-
{name: "should_copy_the_existing_checksum", fn: integration.CopyObject_should_copy_the_existing_checksum},
190-
{name: "should_replace_the_existing_checksum", fn: integration.CopyObject_should_replace_the_existing_checksum},
191-
{name: "to_itself_by_replacing_the_checksum", fn: integration.CopyObject_to_itself_by_replacing_the_checksum},
192192
{name: "incorrect_source_bucket_expected_owner", fn: integration.CopyObject_incorrect_source_bucket_expected_owner},
193193
}
194194

‎logstore/config.go‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -60,8 +60,12 @@ type PlaneConfig struct {
6060
}
6161

6262
// FlushFunc is the contract for shipping one sealed segment's CAR to
63-
// Forge. The segment is single-plane, so no plane argument is needed.
64-
type FlushFunc func(ctx context.Context, seg *Segment) error
63+
// Forge. The segment is single-plane, so no plane argument is needed. On
64+
// success it returns the digest of the shipped sharded-dag-index blob (the
65+
// ship registers the CAR and its index in the bucket's space; DeleteBucket
66+
// releases both), or nil when nothing registered (header-only segment, or a
67+
// non-publishing uploader).
68+
type FlushFunc func(ctx context.Context, seg *Segment) ([]byte, error)
6569

6670
func (c *Config) validate() error {
6771
if c.Dir == "" {

‎logstore/manager.go‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111

1212
block "github.com/ipfs/go-block-format"
1313
"github.com/ipfs/go-cid"
14+
"github.com/multiformats/go-multihash"
1415
"go.uber.org/zap"
1516

1617
"github.com/fil-forge/ingot/blockstore"
@@ -219,6 +220,63 @@ func (m *Manager) Close(ctx context.Context) error {
219220
return nil
220221
}
221222

223+
// QuiesceBucketLog stops bucket's flush pipeline and waits for any in-flight
224+
// ship to finish, so the segment rows ShippedSegmentDigests reads afterwards
225+
// are final: a ship that was mid-flight has either completed (registered its
226+
// blobs AND stamped shipped_at + index_digest) or aborted. Without this,
227+
// DeleteBucket races the flush goroutine — a segment whose CAR just
228+
// registered but wasn't yet marked shipped would be invisible to the release
229+
// pass, and the space delete would be refused. The closed store reopens
230+
// lazily on the bucket's next use, so a delete that fails downstream leaves
231+
// the bucket functional.
232+
func (m *Manager) QuiesceBucketLog(ctx context.Context, bucket string) error {
233+
if err := validBucketDir(bucket); err != nil {
234+
return err
235+
}
236+
m.mu.Lock()
237+
s, ok := m.stores[bucket]
238+
delete(m.stores, bucket)
239+
m.mu.Unlock()
240+
if !ok {
241+
return nil
242+
}
243+
if err := s.Close(ctx); err != nil {
244+
return fmt.Errorf("logstore: manager: quiesce log for %q: %w", bucket, err)
245+
}
246+
return nil
247+
}
248+
249+
// ShippedSegmentDigests returns the multihash of every blob the bucket's
250+
// catalog segments may have registered in its space: each shipped segment's
251+
// CAR and its sharded-dag-index blob, plus the CAR of any sealed-but-
252+
// unshipped segment (a flush aborted between the CAR's blob/add and the
253+
// shipped stamp leaves that registration behind; releasing an unregistered
254+
// blob is a no-op, so over-listing is safe). DeleteBucket must release them
255+
// all before the space itself can be deleted — the tenant service refuses to
256+
// delete a space that still holds registrations. Call QuiesceBucketLog first
257+
// so no ship is in flight while this reads.
258+
func (m *Manager) ShippedSegmentDigests(ctx context.Context, bucket string) ([][]byte, error) {
259+
rows, err := m.meta.ListSegments(ctx, blockstore.PlaneCatalog, bucket)
260+
if err != nil {
261+
return nil, fmt.Errorf("logstore: manager: list segments for %q: %w", bucket, err)
262+
}
263+
var out [][]byte
264+
for _, r := range rows {
265+
if r.State != StateSealed || len(r.SHA256) == 0 {
266+
continue
267+
}
268+
carDigest, err := multihash.Encode(r.SHA256, multihash.SHA2_256)
269+
if err != nil {
270+
return nil, fmt.Errorf("logstore: manager: encode segment %d sha: %w", r.Seq, err)
271+
}
272+
out = append(out, carDigest)
273+
if r.ShippedAt != 0 && len(r.IndexDigest) > 0 {
274+
out = append(out, r.IndexDigest)
275+
}
276+
}
277+
return out, nil
278+
}
279+
222280
// RemoveBucketLog deletes bucket's log entirely: closes its store (dropping
223281
// queued-but-unshipped segments — a deleted bucket's history has nowhere to
224282
// ship), unlinks its directory, and removes its segment rows. Used by

0 commit comments

Comments
 (0)