feat: deferred-accept multipart (FIL-520) + network blob removal on DeleteObject (FIL-588)

blockstore/spool.go

@@ -50,6 +50,18 @@ func (s *Spool) Path(digest mh.Multihash) string {
 	return filepath.Join(s.dir, hex.EncodeToString(digest))
 }
 
+// Remove deletes the blob with the given digest from the spool. Idempotent:
+// removing a blob that isn't spooled is not an error. Callers own the
+// is-it-safe-to-delete question (shared, content-addressed blobs may be
+// referenced by other parts or committed objects).
+func (s *Spool) Remove(digest mh.Multihash) error {
+	err := os.Remove(s.Path(digest))
+	if err != nil && !errors.Is(err, os.ErrNotExist) {
+		return fmt.Errorf("blockstore: spool remove: %w", err)
+	}
+	return nil
+}
+
 // WriteBlob streams r to the spool, computing its sha256 digest as it writes so
 // the blob is never held whole in memory (object-body blobs run up to
 // max_blob_size = 256 MiB; buffering them would put that × concurrency in RAM).

config/config.go

@@ -76,6 +76,12 @@ type Config struct {
 	// may authorize registered provider DIDs directly).
 	AuthServiceProofs string `mapstructure:"auth_service_proofs" yaml:"auth_service_proofs"`
 
+	// MultipartSessionTTL bounds abandoned multipart uploads (Go duration
+	// string, e.g. "168h"): open sessions older than this are aborted by a
+	// background sweeper and their spooled parts dropped. Empty → default
+	// 7 days; a negative duration disables the sweeper.
+	MultipartSessionTTL string `mapstructure:"multipart_session_ttl" yaml:"multipart_session_ttl"`
+
 	// CatalogPlane overrides the catalog logstore pipeline knobs. Any field
 	// left zero/unset falls back to the top-level SealBytes / SealAge / Retain
 	// (and Ship defaults to true) — e.g. to configure the catalog never to ship.
@@ -125,6 +131,13 @@ func (c Config) ServerConfig() (ServerConfig, error) {
 	if err != nil {
 		return ServerConfig{}, err
 	}
+	var mpTTL time.Duration
+	if c.MultipartSessionTTL != "" {
+		mpTTL, err = time.ParseDuration(c.MultipartSessionTTL)
+		if err != nil {
+			return ServerConfig{}, fmt.Errorf("ingot: parse multipart_session_ttl %q: %w", c.MultipartSessionTTL, err)
+		}
+	}
 	// Render the CORS configuration here — the single place it is built —
 	// so a typo fails at startup (via Validate) rather than from New.
 	corsCfg, err := cors.Build(c.CORSAllowedOrigins)
@@ -148,6 +161,8 @@ func (c Config) ServerConfig() (ServerConfig, error) {
 		SealAgeCatalog:   catAge,
 		ShipCatalog:      shipDefault(c.CatalogPlane.Ship),
 		RetainCatalog:    firstNonZeroInt(c.CatalogPlane.Retain, c.Retain),
+
+		MultipartSessionTTL: mpTTL,
 	}, nil
 }
 

config/server.go

@@ -46,6 +46,13 @@ type ServerConfig struct {
 	MaxConnections int
 	MaxRequests    int
 
+	// MultipartSessionTTL bounds abandoned multipart uploads: open sessions
+	// older than this are aborted by a background sweeper (dropping their
+	// spooled parts), and completed-session rows retained for Complete
+	// idempotency are reaped past the same age. Zero → default 7 days;
+	// negative → sweeper disabled.
+	MultipartSessionTTL time.Duration
+
 	// CORSConfig is the S3 CORS configuration the backend reports for
 	// every bucket, rendered from Config.CORSAllowedOrigins by
 	// internal/cors. Nil disables CORS entirely (the default).

forgeclient/blobabort.go

@@ -0,0 +1,58 @@
+package forgeclient
+
+import (
+	"context"
+	"fmt"
+
+	blobcmds "github.com/fil-forge/libforge/commands/blob"
+	ucanlib "github.com/fil-forge/libforge/ucan"
+	"github.com/fil-forge/ucantone/did"
+	"github.com/fil-forge/ucantone/execution"
+	"github.com/fil-forge/ucantone/ucan/invocation"
+	"github.com/ipfs/go-cid"
+	"github.com/multiformats/go-multihash"
+)
+
+// BlobAbort invokes /blob/abort against the upload service (sprue),
+// abandoning the space's in-flight upload of a parked (never-accepted)
+// blob. cause is the /blob/add task link (AddedBlob.AddTask) — sprue
+// walks its receipt chain to locate the storage node holding the parked
+// bytes (which have no registration or acceptance to look up by) and
+// forwards a /blob/reject there. The space is the invocation subject.
+// Idempotent on the node; a blob the space has accepted is refused with
+// BlobAccepted (release it via the reference index / /blob/remove instead).
+func (c *Client) BlobAbort(ctx context.Context, space did.DID, digest multihash.Multihash, cause cid.Cid, options ...BlobAddOption) error {
+	cfg := NewBlobAddConfig(options...)
+	proofStore := ucanlib.ProofStore(c.tokenStore)
+	if cfg.ProofStore != nil {
+		proofStore = cfg.ProofStore
+	}
+
+	proofs, proofLinks, err := proofStore.ProofChain(ctx, c.signer.DID(), blobcmds.Abort.Command, space)
+	if err != nil {
+		return fmt.Errorf("building proof chain: %w", err)
+	}
+	inv, err := blobcmds.Abort.Invoke(
+		c.signer,
+		space,
+		// The space is the invocation subject; it is not repeated in the
+		// arguments.
+		&blobcmds.AbortArguments{Digest: digest, Cause: cause},
+		invocation.WithAudience(c.serviceID),
+		invocation.WithProofs(proofLinks...),
+	)
+	if err != nil {
+		return fmt.Errorf("creating invocation: %w", err)
+	}
+
+	_, _, _, err = Execute[*blobcmds.AbortOK](
+		ctx,
+		c.ucanClient,
+		inv,
+		execution.WithDelegations(proofs...),
+	)
+	if err != nil {
+		return fmt.Errorf("executing invocation: %w", err)
+	}
+	return nil
+}

forgeclient/blobadd.go

@@ -6,6 +6,9 @@
 //     not just the client's token store.
 //   - No /blob/accept re-delegation: sprue owns accept (as it owns
 //     allocate), so the conclude/put-receipt dance carries no space proof.
+//   - BlobAdd accepts WithConclude(false) so multipart can defer the
+//     conclude (park at UploadPart, accept at Complete); BlobConclude
+//     finishes the parked add later.
 //
 // Also dropped from upstream: otel spans, go-log, ctxutil, the progress/stall
 // readers, and the hard requirement that the accept receipt carry a PDP
@@ -55,11 +58,14 @@ type BlobAddConfig struct {
 	// call instead of the client's default token store — used to scope an
 	// invocation to a request's per-access-key proofs.
 	ProofStore ucanlib.ProofStore
+	// Conclude controls whether BlobAdd concludes the /http/put receipt
+	// (triggering /blob/accept) before returning. Default true.
+	Conclude bool
 }
 
 // NewBlobAddConfig builds a BlobAddConfig from options.
 func NewBlobAddConfig(options ...BlobAddOption) *BlobAddConfig {
-	cfg := &BlobAddConfig{PutClient: &http.Client{}}
+	cfg := &BlobAddConfig{PutClient: &http.Client{}, Conclude: true}
 	for _, opt := range options {
 		opt(cfg)
 	}
@@ -86,20 +92,65 @@ func WithProofStore(ps ucanlib.ProofStore) BlobAddOption {
 	return func(cfg *BlobAddConfig) { cfg.ProofStore = ps }
 }
 
-// AddedBlob is the result of a successful BlobAdd.
+// WithConclude controls whether BlobAdd concludes the upload before
+// returning. WithConclude(false) leaves the blob parked — durable on the
+// provider, but piri holds the bytes without aggregating them until
+// /blob/accept fires: the returned AddedBlob has a nil Location and carries
+// the state [Client.BlobConclude] needs to finish the upload later (or
+// [Client.BlobAbort] to abandon it; AddTask is the abort Cause). Moot on
+// dedup — when the provider already held accepted bytes for the content,
+// accept already ran and the add completes regardless.
+func WithConclude(conclude bool) BlobAddOption {
+	return func(cfg *BlobAddConfig) { cfg.Conclude = conclude }
+}
+
+// AddedBlob is the result of a BlobAdd. Location is set once the blob is
+// accepted; with WithConclude(false) it is nil until the deferred
+// [Client.BlobConclude] — persist the task links + PutInvocation in between.
 type AddedBlob struct {
-	Digest   multihash.Multihash
-	Size     uint64
-	Location ucan.Invocation // the /assert/location commitment
+	Digest multihash.Multihash
+	Size   uint64
+	// Location is the /assert/location commitment issued at accept; nil
+	// while the add is unconcluded (parked).
+	Location ucan.Invocation
+	// AddTask is the /blob/add task link — the receipt-chain root the
+	// upload service uses to locate the provider for abort.
+	AddTask cid.Cid
+	// AcceptTask is the /blob/accept task link BlobConclude polls.
+	AcceptTask cid.Cid
+	// PutInvocation is the issued /http/put invocation, populated only while
+	// the add is unconcluded. Its metadata embeds the derived signer keys
+	// needed to synthesize the put receipt at conclude time — treat it as
+	// sensitive and delete it once concluded or rejected.
+	PutInvocation []byte
 }
 
 // BlobAdd adds a blob to the upload service (sprue): invoke /blob/add,
 // PUT the bytes, conclude a synthesized /http/put receipt, then poll the
 // /blob/accept receipt for the location commitment. The issuer needs a
-// /blob/add delegation proof over space.
-func (c *Client) BlobAdd(ctx context.Context, space did.DID, content io.Reader, options ...BlobAddOption) (blob AddedBlob, err error) {
+// /blob/add delegation proof over space. With WithConclude(false) it stops
+// after the PUT — the blob stays parked until [Client.BlobConclude].
+func (c *Client) BlobAdd(ctx context.Context, space did.DID, content io.Reader, options ...BlobAddOption) (AddedBlob, error) {
 	cfg := NewBlobAddConfig(options...)
+	added, err := c.blobAdd(ctx, space, content, cfg)
+	if err != nil {
+		return AddedBlob{}, err
+	}
+	// Already accepted (dedup) or deliberately unconcluded — done either way.
+	if added.Location != nil || !cfg.Conclude {
+		return added, nil
+	}
+	return c.BlobConclude(ctx, space, added)
+}
 
+// blobAdd runs the durable half of BlobAdd: /blob/add + PUT the bytes,
+// WITHOUT concluding the /http/put receipt — the conclude is what makes the
+// upload service trigger /blob/accept on the provider, so the blob stays
+// parked (stored, unaggregated) until BlobConclude. The result's Location is
+// nil unless the provider already held accepted bytes for this content
+// (dedup: allocate returned no upload address and the put receipt was
+// pre-issued, so accept already ran).
+func (c *Client) blobAdd(ctx context.Context, space did.DID, content io.Reader, cfg *BlobAddConfig) (blob AddedBlob, err error) {
 	putClient := cfg.PutClient
 	contentReader := content
 	contentHash := cfg.PrecomputedDigest
@@ -111,7 +162,7 @@ func (c *Client) BlobAdd(ctx context.Context, space did.DID, content io.Reader,
 		if err != nil {
 			c.logger.Error("blob add failed", zap.Stringer("space", space), zap.Error(err), zap.Duration("duration", time.Since(start)))
 		} else {
-			c.logger.Debug("blob added", zap.Stringer("space", space), zap.Duration("duration", time.Since(start)))
+			c.logger.Debug("blob added", zap.Stringer("space", space), zap.Bool("parked", blob.Location == nil), zap.Duration("duration", time.Since(start)))
 		}
 	}()
 
@@ -213,31 +264,95 @@ func (c *Client) BlobAdd(ctx context.Context, space did.DID, content io.Reader,
 		}
 	}
 
-	// Conclude a synthesized /http/put receipt so /blob/accept can resolve.
-	// Accept is owned by sprue (like allocate), so no /blob/accept
-	// re-delegation is attached — the conclude is issued agent→sprue and
-	// carries no space proof.
-	if !putSuccess {
-		if err := c.sendPutReceipt(ctx, putInv); err != nil {
-			return AddedBlob{}, fmt.Errorf("sending put receipt: %w", err)
+	// Dedup path: the provider already held accepted bytes for this content,
+	// so the put receipt was pre-issued and the upload service ran accept
+	// synchronously — the blob is not parked. Await the accept receipt and
+	// return the completed AddedBlob.
+	if putSuccess {
+		location, aerr := c.awaitAccept(ctx, accInv.Task().Link())
+		if aerr != nil {
+			return AddedBlob{}, aerr
 		}
+		return AddedBlob{
+			Digest:     contentHash,
+			Size:       *contentSizePtr,
+			Location:   location,
+			AddTask:    inv.Task().Link(),
+			AcceptTask: accInv.Task().Link(),
+		}, nil
+	}
+
+	// Parked: durable on the provider, conclude deferred to BlobConclude.
+	return AddedBlob{
+		Digest:        contentHash,
+		Size:          *contentSizePtr,
+		AddTask:       inv.Task().Link(),
+		AcceptTask:    accInv.Task().Link(),
+		PutInvocation: putInv.Bytes(),
+	}, nil
+}
+
+// BlobConclude finishes a parked (unconcluded) BlobAdd: it synthesizes and
+// concludes the /http/put receipt (which makes the upload service trigger
+// /blob/accept on the provider) and awaits the accept receipt's location
+// commitment. Accept is owned by sprue (like allocate), so the conclude
+// carries no space proof. Safe to retry — re-concluding an already-concluded
+// put is tolerated upstream, and an AddedBlob whose Location is already set
+// returns as-is. The result drops PutInvocation (spent — the caller should
+// delete its persisted copy too).
+func (c *Client) BlobConclude(ctx context.Context, space did.DID, added AddedBlob) (blob AddedBlob, err error) {
+	if added.Location != nil {
+		return added, nil
+	}
+	start := time.Now()
+	defer func() {
+		if err != nil {
+			c.logger.Error("blob conclude failed", zap.Stringer("space", space), zap.Error(err), zap.Duration("duration", time.Since(start)))
+		} else {
+			c.logger.Debug("blob concluded", zap.Stringer("space", space), zap.Duration("duration", time.Since(start)))
+		}
+	}()
+
+	putInv := new(invocation.Invocation)
+	if err := putInv.UnmarshalCBOR(bytes.NewReader(added.PutInvocation)); err != nil {
+		return AddedBlob{}, fmt.Errorf("decoding parked /http/put invocation: %w", err)
+	}
+
+	if err := c.sendPutReceipt(ctx, putInv); err != nil {
+		return AddedBlob{}, fmt.Errorf("sending put receipt: %w", err)
 	}
 
-	accRcpt, accMeta, err := c.receiptsClient.Poll(ctx, accInv.Task().Link(), receipt_client.WithRetries(5))
+	location, err := c.awaitAccept(ctx, added.AcceptTask)
 	if err != nil {
-		return AddedBlob{}, fmt.Errorf("polling accept receipt: %w", err)
+		return AddedBlob{}, err
+	}
+	return AddedBlob{
+		Digest:     added.Digest,
+		Size:       added.Size,
+		Location:   location,
+		AddTask:    added.AddTask,
+		AcceptTask: added.AcceptTask,
+	}, nil
+}
+
+// awaitAccept polls the /blob/accept receipt and extracts the
+// /assert/location commitment from its metadata.
+func (c *Client) awaitAccept(ctx context.Context, acceptTask cid.Cid) (ucan.Invocation, error) {
+	accRcpt, accMeta, err := c.receiptsClient.Poll(ctx, acceptTask, receipt_client.WithRetries(5))
+	if err != nil {
+		return nil, fmt.Errorf("polling accept receipt: %w", err)
 	}
-	o, x = accRcpt.Out().Unpack()
+	o, x := accRcpt.Out().Unpack()
 	if accRcpt.Out().IsErr() {
 		var model edm.ErrorModel
 		if err := model.UnmarshalCBOR(bytes.NewReader(x)); err != nil {
-			return AddedBlob{}, fmt.Errorf("executing invocation")
+			return nil, fmt.Errorf("executing invocation")
 		}
-		return AddedBlob{}, fmt.Errorf("failure in accept receipt: %w", model)
+		return nil, fmt.Errorf("failure in accept receipt: %w", model)
 	}
 	var accOK blobcmds.AcceptOK
 	if err := accOK.UnmarshalCBOR(bytes.NewReader(o)); err != nil {
-		return AddedBlob{}, fmt.Errorf("unmarshaling accept receipt output: %w", err)
+		return nil, fmt.Errorf("unmarshaling accept receipt output: %w", err)
 	}
 
 	var locationCommitment ucan.Invocation
@@ -247,10 +362,9 @@ func (c *Client) BlobAdd(ctx context.Context, space did.DID, content io.Reader,
 		}
 	}
 	if locationCommitment == nil {
-		return AddedBlob{}, fmt.Errorf("blob accept receipt missing location commitment invocation")
+		return nil, fmt.Errorf("blob accept receipt missing location commitment invocation")
 	}
-
-	return AddedBlob{Digest: contentHash, Size: *contentSizePtr, Location: locationCommitment}, nil
+	return locationCommitment, nil
 }
 
 func putBlob(ctx context.Context, client *http.Client, url *url.URL, headers map[string]string, body io.Reader, size int64) error {