Skip to content

feat(s3): multipart checksums (FIL-620) #68

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
frrist merged 4 commits into from
Aug 10, 2026
Merged

feat(s3): multipart checksums (FIL-620) #68

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
948258d
feat(s3): multipart checksums (FIL-620)
frrist Aug 6, 2026
6bb50d8
feat(s3): CopyObject checksum propagation and replacement
frrist Aug 6, 2026
4b7e11f
fix(s3): release shipped segment blobs on DeleteBucket
frrist Aug 6, 2026
c3deaf3
Update bucket/manifest.go comment
frrist Aug 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions blockstore/staging_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ func (f *fakeMeta) MarkSegmentSealed(_ context.Context, plane blockstore.Plane,
f.roots = append(f.roots, opRoots...)
return nil
}
func (f *fakeMeta) MarkSegmentShipped(_ context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, _ []blockstore.OpRoot) error {
func (f *fakeMeta) MarkSegmentShipped(_ context.Context, plane blockstore.Plane, seq uint64, shippedAt int64, _ []byte, _ []blockstore.OpRoot) error {
if r, ok := f.rows[seq]; ok {
r.ShippedAt = shippedAt
}
Expand Down Expand Up @@ -86,7 +86,7 @@ func (f *fakeMeta) RehydrateSegment(_ context.Context, m logstore.SegmentMeta) e

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

// noopBase satisfies blockstore.BlockReader but always returns
// errUnknownBase so we can detect when a GetBlock falls through
Expand Down
36 changes: 35 additions & 1 deletion bucket/cbor_gen.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

5 changes: 5 additions & 0 deletions bucket/manifest.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,11 @@ type ObjectManifest struct {
// checksum was requested. Independent of the internal sha256 content address.
ChecksumAlgorithm string `cborgen:"ca"`
Checksum string `cborgen:"ck"`
// ChecksumType is the S3 checksum type of Checksum: "FULL_OBJECT" (computed
// over the whole body) or "COMPOSITE" (a multipart checksum-of-checksums
// with a "-N" part-count suffix). Empty in blocks written before the type
// was recorded, which are all full-object.
ChecksumType string `cborgen:"cy"`

// HTTP/S3 system headers carried through PUT and replayed on
// HEAD/GET. Empty strings are omitted from responses.
Expand Down
4 changes: 2 additions & 2 deletions inmem/segments_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ func TestMarkSegmentShipped_GuardsForgeRootOnRoot(t *testing.T) {

// Ship with the stale op-root LAST: unconditionally (the old behavior) it
// would win as the last write; the guard must skip it and keep `committed`.
if err := m.MarkSegmentShipped(ctx, blockstore.PlaneCatalog, seq, 100, []blockstore.OpRoot{
if err := m.MarkSegmentShipped(ctx, blockstore.PlaneCatalog, seq, 100, nil, []blockstore.OpRoot{
{Bucket: "bk", Root: committed},
{Bucket: "bk", Root: stale},
}); err != nil {
Expand All @@ -65,7 +65,7 @@ func TestMarkSegmentShipped_GuardsForgeRootOnRoot(t *testing.T) {
_ = m2.CASRoot(ctx, "bk2", cid.Undef, committed)
seq2, _ := m2.NextSegmentSeq(ctx)
_ = m2.InsertSegmentOpen(ctx, blockstore.PlaneCatalog, seq2, "bk2")
if err := m2.MarkSegmentShipped(ctx, blockstore.PlaneCatalog, seq2, 100, []blockstore.OpRoot{
if err := m2.MarkSegmentShipped(ctx, blockstore.PlaneCatalog, seq2, 100, nil, []blockstore.OpRoot{
{Bucket: "bk2", Root: stale},
}); err != nil {
t.Fatalf("MarkSegmentShipped(stale only): %v", err)
Expand Down
8 changes: 5 additions & 3 deletions inmem/store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ import (
"fmt"
"io"
"net/url"
"slices"
"sort"
"strconv"
"strings"
Expand Down Expand Up @@ -287,11 +288,12 @@ func (m *MemStore) MarkSegmentSealed(_ context.Context, plane blockstore.Plane,
return nil
}

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

func (NopUploader) SubmitShard(_ context.Context, _ blockstore.Plane, _ did.DID, _ uploader.CARShard) (uploader.BlobLocation, error) {
return uploader.BlobLocation{}, nil
func (NopUploader) SubmitShard(_ context.Context, _ blockstore.Plane, _ did.DID, _ uploader.CARShard) (uploader.BlobLocation, multihash.Multihash, error) {
return uploader.BlobLocation{}, nil, nil
}

// UploadBlob accepts immediately, even with WithConclude(false) — there is
Expand Down
45 changes: 21 additions & 24 deletions itest/versity_multipart_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,9 @@ import (

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

var createMultipartPass = []forgeCase{
{name: "non_existing_bucket", fn: integration.CreateMultipartUpload_non_existing_bucket},
Expand Down Expand Up @@ -45,18 +46,17 @@ var uploadPartPass = []forgeCase{
{name: "multiple_checksum_headers", fn: integration.UploadPart_multiple_checksum_headers},
{name: "invalid_checksum_header", fn: integration.UploadPart_invalid_checksum_header},
{name: "checksum_header_and_algo_mismatch", fn: integration.UploadPart_checksum_header_and_algo_mismatch},
{name: "success", fn: integration.UploadPart_success},
}

var uploadPartXFail = []forgeCase{
{name: "checksum_algorithm_mistmatch_on_initialization", fn: integration.UploadPart_checksum_algorithm_mistmatch_on_initialization},
{name: "checksum_algorithm_mistmatch_on_initialization_with_value", fn: integration.UploadPart_checksum_algorithm_mistmatch_on_initialization_with_value},
{name: "incorrect_checksums", fn: integration.UploadPart_incorrect_checksums},
{name: "no_checksum_with_full_object_checksum_type", fn: integration.UploadPart_no_checksum_with_full_object_checksum_type},
{name: "no_checksum_with_composite_checksum_type", fn: integration.UploadPart_no_checksum_with_composite_checksum_type},
{name: "with_checksums_success", fn: integration.UploadPart_with_checksums_success},
{name: "success", fn: integration.UploadPart_success},
}

var uploadPartXFail = []forgeCase{}

var uploadPartCopyPass = []forgeCase{
{name: "non_existing_bucket", fn: integration.UploadPartCopy_non_existing_bucket},
{name: "invalid_part_number", fn: integration.UploadPartCopy_invalid_part_number},
Expand Down Expand Up @@ -90,13 +90,11 @@ var listPartsPass = []forgeCase{
{name: "truncated", fn: integration.ListParts_truncated},
{name: "success", fn: integration.ListParts_success},
{name: "with_checksums", fn: integration.ListParts_with_checksums},
}

// Explicit null-checksum-type echo is FIL-620.
var listPartsXFail = []forgeCase{
{name: "null_checksums", fn: integration.ListParts_null_checksums},
}

var listPartsXFail = []forgeCase{}

var listMultipartUploadsPass = []forgeCase{
{name: "non_existing_bucket", fn: integration.ListMultipartUploads_non_existing_bucket},
{name: "empty_result", fn: integration.ListMultipartUploads_empty_result},
Expand Down Expand Up @@ -142,6 +140,18 @@ var completeMultipartPass = []forgeCase{
{name: "multiple_final_checksums", fn: integration.CompleteMultipartUpload_multiple_final_checksums},
{name: "invalid_final_checksums", fn: integration.CompleteMultipartUpload_invalid_final_checksums},
{name: "invalid_final_composite_checksum", fn: integration.CompleteMultipartUpload_invalid_final_composite_checksum},
{name: "invalid_checksum_part", fn: integration.CompleteMultipartUpload_invalid_checksum_part},
{name: "multiple_checksum_part", fn: integration.CompleteMultipartUpload_multiple_checksum_part},
{name: "incorrect_checksum_part", fn: integration.CompleteMultipartUpload_incorrect_checksum_part},
{name: "different_checksum_part", fn: integration.CompleteMultipartUpload_different_checksum_part},
{name: "missing_part_checksum", fn: integration.CompleteMultipartUpload_missing_part_checksum},
{name: "incorrect_final_checksums", fn: integration.CompleteMultipartUpload_incorrect_final_checksums},
{name: "should_calculate_the_final_checksum_full_object", fn: integration.CompleteMultipartUpload_should_calculate_the_final_checksum_full_object},
{name: "should_verify_the_final_checksum", fn: integration.CompleteMultipartUpload_should_verify_the_final_checksum},
{name: "should_verify_final_composite_checksum", fn: integration.CompleteMultipartUpload_should_verify_final_composite_checksum},
{name: "checksum_type_mismatch", fn: integration.CompleteMultipartUpload_checksum_type_mismatch},
{name: "should_ignore_the_final_checksum", fn: integration.CompleteMultipartUpload_should_ignore_the_final_checksum},
{name: "should_succeed_without_final_checksum_type", fn: integration.CompleteMultipartUpload_should_succeed_without_final_checksum_type},
{name: "with_metadata", fn: integration.CompleteMultipartUpload_with_metadata},
{name: "success", fn: integration.CompleteMultipartUpload_success},
{name: "already_completed", fn: integration.CompleteMultipartUpload_already_completed},
Expand All @@ -158,20 +168,7 @@ var completeMultipartPass = []forgeCase{
}},
}

// Part-level / composite-checksum verification is FIL-620;
// racey_data_integrity additionally leans on atomic concurrent overwrites.
// racey_data_integrity leans on atomic concurrent overwrites under load.
var completeMultipartXFail = []forgeCase{
{name: "invalid_checksum_part", fn: integration.CompleteMultipartUpload_invalid_checksum_part},
{name: "multiple_checksum_part", fn: integration.CompleteMultipartUpload_multiple_checksum_part},
{name: "incorrect_checksum_part", fn: integration.CompleteMultipartUpload_incorrect_checksum_part},
{name: "different_checksum_part", fn: integration.CompleteMultipartUpload_different_checksum_part},
{name: "missing_part_checksum", fn: integration.CompleteMultipartUpload_missing_part_checksum},
{name: "incorrect_final_checksums", fn: integration.CompleteMultipartUpload_incorrect_final_checksums},
{name: "should_calculate_the_final_checksum_full_object", fn: integration.CompleteMultipartUpload_should_calculate_the_final_checksum_full_object},
{name: "should_verify_the_final_checksum", fn: integration.CompleteMultipartUpload_should_verify_the_final_checksum},
{name: "should_verify_final_composite_checksum", fn: integration.CompleteMultipartUpload_should_verify_final_composite_checksum},
{name: "checksum_type_mismatch", fn: integration.CompleteMultipartUpload_checksum_type_mismatch},
{name: "should_ignore_the_final_checksum", fn: integration.CompleteMultipartUpload_should_ignore_the_final_checksum},
{name: "should_succeed_without_final_checksum_type", fn: integration.CompleteMultipartUpload_should_succeed_without_final_checksum_type},
{name: "racey_data_integrity", fn: integration.CompleteMultipartUpload_racey_data_integrity},
}
10 changes: 5 additions & 5 deletions itest/versity_object_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -173,10 +173,14 @@ var copyObjectPass = []forgeCase{
{name: "invalid_legal_hold", fn: integration.CopyObject_invalid_legal_hold},
{name: "invalid_object_lock_mode", fn: integration.CopyObject_invalid_object_lock_mode},
{name: "invalid_website_redirect_location", fn: integration.CopyObject_invalid_website_redirect_location},
{name: "create_checksum_on_copy", fn: integration.CopyObject_create_checksum_on_copy},
{name: "should_copy_the_existing_checksum", fn: integration.CopyObject_should_copy_the_existing_checksum},
{name: "should_replace_the_existing_checksum", fn: integration.CopyObject_should_replace_the_existing_checksum},
{name: "to_itself_by_replacing_the_checksum", fn: integration.CopyObject_to_itself_by_replacing_the_checksum},
}

// Observed failing against the forge stack: multi-account semantics, tagging,
// object-lock, and checksum-on-copy are unimplemented surface.
// and object-lock are unimplemented surface.
var copyObjectXFail = []forgeCase{
{name: "not_owned_source_bucket", fn: integration.CopyObject_not_owned_source_bucket},
{name: "should_replace_tagging", fn: integration.CopyObject_should_replace_tagging},
Expand All @@ -185,10 +189,6 @@ var copyObjectXFail = []forgeCase{
{name: "with_legal_hold", fn: integration.CopyObject_with_legal_hold},
{name: "with_retention_lock", fn: integration.CopyObject_with_retention_lock},
{name: "object_acl_not_supported", fn: integration.CopyObject_object_acl_not_supported},
{name: "create_checksum_on_copy", fn: integration.CopyObject_create_checksum_on_copy},
{name: "should_copy_the_existing_checksum", fn: integration.CopyObject_should_copy_the_existing_checksum},
{name: "should_replace_the_existing_checksum", fn: integration.CopyObject_should_replace_the_existing_checksum},
{name: "to_itself_by_replacing_the_checksum", fn: integration.CopyObject_to_itself_by_replacing_the_checksum},
{name: "incorrect_source_bucket_expected_owner", fn: integration.CopyObject_incorrect_source_bucket_expected_owner},
}

Expand Down
8 changes: 6 additions & 2 deletions logstore/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,8 +60,12 @@ type PlaneConfig struct {
}

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

func (c *Config) validate() error {
if c.Dir == "" {
Expand Down
58 changes: 58 additions & 0 deletions logstore/manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (

block "github.com/ipfs/go-block-format"
"github.com/ipfs/go-cid"
"github.com/multiformats/go-multihash"
"go.uber.org/zap"

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

// QuiesceBucketLog stops bucket's flush pipeline and waits for any in-flight
// ship to finish, so the segment rows ShippedSegmentDigests reads afterwards
// are final: a ship that was mid-flight has either completed (registered its
// blobs AND stamped shipped_at + index_digest) or aborted. Without this,
// DeleteBucket races the flush goroutine — a segment whose CAR just
// registered but wasn't yet marked shipped would be invisible to the release
// pass, and the space delete would be refused. The closed store reopens
// lazily on the bucket's next use, so a delete that fails downstream leaves
// the bucket functional.
func (m *Manager) QuiesceBucketLog(ctx context.Context, bucket string) error {
if err := validBucketDir(bucket); err != nil {
return err
}
m.mu.Lock()
s, ok := m.stores[bucket]
delete(m.stores, bucket)
m.mu.Unlock()
if !ok {
return nil
}
if err := s.Close(ctx); err != nil {
return fmt.Errorf("logstore: manager: quiesce log for %q: %w", bucket, err)
}
return nil
}

// ShippedSegmentDigests returns the multihash of every blob the bucket's
// catalog segments may have registered in its space: each shipped segment's
// CAR and its sharded-dag-index blob, plus the CAR of any sealed-but-
// unshipped segment (a flush aborted between the CAR's blob/add and the
// shipped stamp leaves that registration behind; releasing an unregistered
// blob is a no-op, so over-listing is safe). DeleteBucket must release them
// all before the space itself can be deleted — the tenant service refuses to
// delete a space that still holds registrations. Call QuiesceBucketLog first
// so no ship is in flight while this reads.
func (m *Manager) ShippedSegmentDigests(ctx context.Context, bucket string) ([][]byte, error) {

@alanshaw alanshaw Aug 9, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should this be an iterator? It could be big in buckets that have lived for a long time right? We'll probably find we need a bulk /blob/remove or GC beforehand somehow to allow this to complete in a reasonable time.

@frrist frrist Aug 10, 2026 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It stays small by construction, so an iterator would not buy anything: retention prunes segment rows past the Retain window (default 6), so this enumerates roughly Retain shipped segments plus a handful of unshipped tails, regardless of the bucket's age.

The pruning is also the sharp edge you are gesturing at. A retired segment loses its row here, but its CAR and index stay registered in the space — the network copy is the durable read tier, so releasing at retirement would be wrong. That means this release pass only covers registrations the rows still describe: a long-lived bucket with more than Retain shipped segments still cannot be fully released from ingot's side, and the per-blob /blob/remove loop would not scale to that count anyway.

So agreed on where this lands: the durable fix is a space-scoped bulk release (or the tenant service tearing down a space it knows is being deleted) rather than ingot enumerating per blob — that would also sweep any registrations leaked by past best-effort release failures. Filed as #76. This pass keeps bucket deletion correct for the window the rows cover, which is also everything the conformance suite exercises.

@frrist frrist Aug 10, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Issue here: #76

rows, err := m.meta.ListSegments(ctx, blockstore.PlaneCatalog, bucket)
if err != nil {
return nil, fmt.Errorf("logstore: manager: list segments for %q: %w", bucket, err)
}
var out [][]byte
for _, r := range rows {
if r.State != StateSealed || len(r.SHA256) == 0 {
continue
}
carDigest, err := multihash.Encode(r.SHA256, multihash.SHA2_256)
if err != nil {
return nil, fmt.Errorf("logstore: manager: encode segment %d sha: %w", r.Seq, err)
}
out = append(out, carDigest)
if r.ShippedAt != 0 && len(r.IndexDigest) > 0 {
out = append(out, r.IndexDigest)
}
}
return out, nil
}

// RemoveBucketLog deletes bucket's log entirely: closes its store (dropping
// queued-but-unshipped segments — a deleted bucket's history has nowhere to
// ship), unlinks its directory, and removes its segment rows. Used by
Expand Down
Loading
Loading