feat: blob removal — serve /blob/remove, /upload/remove, /blob/abort (FIL-522)

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/dynamodb v1.57.3
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.96.4
 	github.com/docker/docker v28.5.2+incompatible
-	github.com/fil-forge/libforge v0.0.0-20260724113901-7fc3b2cec1ef
+	github.com/fil-forge/libforge v0.0.0-20260727220215-5e299c46f62f
 	github.com/fil-forge/ucantone v0.0.0-20260713122829-2662bddc02ab
 	github.com/google/uuid v1.6.0
 	github.com/ipfs/go-cid v0.6.1

go.sum

@@ -92,8 +92,8 @@ github.com/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/
 github.com/ebitengine/purego v0.10.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
-github.com/fil-forge/libforge v0.0.0-20260724113901-7fc3b2cec1ef h1:xgkciShyWdCQ2Pyl2qEwP7JMbsdbHdECj/v6E5+fj6U=
-github.com/fil-forge/libforge v0.0.0-20260724113901-7fc3b2cec1ef/go.mod h1:0kXihIQ4L2uZ00nR5XrZ/Y8Db7Ht/qQNuiWslwMJ95M=
+github.com/fil-forge/libforge v0.0.0-20260727220215-5e299c46f62f h1:QzgMg8GIE4IhgOE/7DBHFU/4T5oU7J4vAKxbUPKJPGA=
+github.com/fil-forge/libforge v0.0.0-20260727220215-5e299c46f62f/go.mod h1:0kXihIQ4L2uZ00nR5XrZ/Y8Db7Ht/qQNuiWslwMJ95M=
 github.com/fil-forge/ucantone v0.0.0-20260713122829-2662bddc02ab h1:Kal5frXuFQiaeumu1DreTbaXmK6DXWfz7al4FFhAh4A=
 github.com/fil-forge/ucantone v0.0.0-20260713122829-2662bddc02ab/go.mod h1:oFY5BfD0bDeodGlbBHh3/nK99MAS93rGXjoQz7s5qgE=
 github.com/filecoin-project/go-data-segment v0.0.1 h1:1wmDxOG4ubWQm3ZC1XI5nCon5qgSq7Ra3Rb6Dbu10Gs=

internal/fx/service/handlers/provider.go

@@ -52,6 +52,14 @@ var Module = fx.Module("service-handlers",
 			handlers.NewBlobListHandler,
 			fx.ResultTags(`group:"ucan_handlers"`),
 		),
+		fx.Annotate(
+			handlers.NewBlobRemoveHandler,
+			fx.ResultTags(`group:"ucan_handlers"`),
+		),
+		fx.Annotate(
+			handlers.NewBlobAbortHandler,
+			fx.ResultTags(`group:"ucan_handlers"`),
+		),
 		// fx.Annotate(
 		// 	handlers.NewBlobReplicateHandler,
 		// 	fx.ResultTags(`group:"ucan_handlers"`),
@@ -76,6 +84,10 @@ var Module = fx.Module("service-handlers",
 			handlers.NewUploadAddHandler,
 			fx.ResultTags(`group:"ucan_handlers"`),
 		),
+		fx.Annotate(
+			handlers.NewUploadRemoveHandler,
+			fx.ResultTags(`group:"ucan_handlers"`),
+		),
 		fx.Annotate(
 			handlers.NewUploadListHandler,
 			fx.ResultTags(`group:"ucan_handlers"`),

pkg/piriclient/client.go

@@ -18,6 +18,7 @@ import (
 	"github.com/fil-forge/ucantone/ucan/invocation"
 	"github.com/fil-forge/ucantone/ucan/promise"
 	"github.com/ipfs/go-cid"
+	"github.com/multiformats/go-multihash"
 	"go.uber.org/zap"
 )
 
@@ -58,7 +59,7 @@ func NewWithClient(piriDID did.DID, issuer ucan.Issuer, client *client.HTTPClien
 // AllocateRequest contains the parameters for a /blob/allocate invocation.
 type AllocateRequest struct {
 	Space  did.DID
-	Digest []byte
+	Digest multihash.Multihash
 	Size   uint64
 	Cause  cid.Cid
 }
@@ -132,7 +133,7 @@ func (c *Client) PiriDID() did.DID {
 // AcceptRequest contains the parameters for a /blob/accept invocation.
 type AcceptRequest struct {
 	Space  did.DID
-	Digest []byte
+	Digest multihash.Multihash
 	Size   uint64
 	Put    cid.Cid // Link to the /http/put task that uploaded the blob
 }
@@ -196,10 +197,146 @@ func (c *Client) AcceptInvocation(ctx context.Context, req *AcceptRequest, proof
 	return inv, prfs, nil
 }
 
+// ReleaseRequest contains the parameters for a /blob/release invocation.
+type ReleaseRequest struct {
+	Space  did.DID
+	Digest multihash.Multihash
+}
+
+// Release sends a /blob/release invocation to the piri node, releasing the
+// space's claim on the blob. Returns the response data, the invocation that
+// was sent, and the receipt from piri. Piri's handler is idempotent, so
+// releasing an already-released blob succeeds.
+func (c *Client) Release(ctx context.Context, req *ReleaseRequest, proofStore ucanlib.ProofStore, options ...invocation.Option) (*blobcmds.ReleaseOK, ucan.Invocation, ucan.Receipt, error) {
+	inv, prfs, err := c.ReleaseInvocation(ctx, req, proofStore, options...)
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("creating release invocation: %w", err)
+	}
+
+	c.logger.Debug("RELEASE invocation created",
+		zap.Stringer("issuer", inv.Issuer()),
+		zap.Stringer("audience", inv.Audience()),
+		zap.Int("proofs", len(prfs)),
+	)
+
+	releaseOK, rcpt, _, err := ucan_client.Execute[*blobcmds.ReleaseOK](
+		ctx,
+		c.client,
+		c.logger,
+		inv,
+		execution.WithDelegations(prfs...),
+	)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	return releaseOK, inv, rcpt, nil
+}
+
+// ReleaseInvocation returns the invocation for the release request.
+func (c *Client) ReleaseInvocation(ctx context.Context, req *ReleaseRequest, proofStore ucanlib.ProofStore, options ...invocation.Option) (ucan.Invocation, []ucan.Delegation, error) {
+	// As with allocate/accept, the proof chain is rooted at the storage
+	// provider, so the subject is the provider DID and the space travels in
+	// the arguments.
+	prfs, prfLinks, err := proofStore.ProofChain(ctx, c.issuer.DID(), blobcmds.Release.Command, c.piriDID)
+	if err != nil {
+		return nil, nil, fmt.Errorf("building proof chain: %w", err)
+	}
+
+	options = slices.Clone(options)
+	options = append(
+		options,
+		invocation.WithAudience(c.piriDID),
+		invocation.WithProofs(prfLinks...),
+	)
+
+	inv, err := blobcmds.Release.Invoke(
+		c.issuer,
+		c.piriDID,
+		&blobcmds.ReleaseArguments{
+			Space:  req.Space,
+			Digest: req.Digest,
+		},
+		options...,
+	)
+	if err != nil {
+		return nil, nil, fmt.Errorf("creating release invocation: %w", err)
+	}
+
+	return inv, prfs, nil
+}
+
+// RejectRequest contains the parameters for a /blob/reject invocation.
+type RejectRequest struct {
+	Space  did.DID
+	Digest multihash.Multihash
+}
+
+// Reject sends a /blob/reject invocation to the piri node, retiring the
+// space's parked (never-accepted) blob. Piri refuses accepted blobs with a
+// BlobAccepted failure; otherwise the handler is idempotent.
+func (c *Client) Reject(ctx context.Context, req *RejectRequest, proofStore ucanlib.ProofStore, options ...invocation.Option) (*blobcmds.RejectOK, ucan.Invocation, ucan.Receipt, error) {
+	inv, prfs, err := c.RejectInvocation(ctx, req, proofStore, options...)
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("creating reject invocation: %w", err)
+	}
+
+	c.logger.Debug("REJECT invocation created",
+		zap.Stringer("issuer", inv.Issuer()),
+		zap.Stringer("audience", inv.Audience()),
+		zap.Int("proofs", len(prfs)),
+	)
+
+	rejectOK, rcpt, _, err := ucan_client.Execute[*blobcmds.RejectOK](
+		ctx,
+		c.client,
+		c.logger,
+		inv,
+		execution.WithDelegations(prfs...),
+	)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	return rejectOK, inv, rcpt, nil
+}
+
+// RejectInvocation returns the invocation for the reject request.
+func (c *Client) RejectInvocation(ctx context.Context, req *RejectRequest, proofStore ucanlib.ProofStore, options ...invocation.Option) (ucan.Invocation, []ucan.Delegation, error) {
+	// As with allocate/accept/release, the proof chain is rooted at the
+	// storage provider, so the subject is the provider DID and the space
+	// travels in the arguments. Cause is not forwarded — it is upload-service
+	// routing metadata, meaningless to the node.
+	prfs, prfLinks, err := proofStore.ProofChain(ctx, c.issuer.DID(), blobcmds.Reject.Command, c.piriDID)
+	if err != nil {
+		return nil, nil, fmt.Errorf("building proof chain: %w", err)
+	}
+
+	options = slices.Clone(options)
+	options = append(
+		options,
+		invocation.WithAudience(c.piriDID),
+		invocation.WithProofs(prfLinks...),
+	)
+
+	inv, err := blobcmds.Reject.Invoke(
+		c.issuer,
+		c.piriDID,
+		&blobcmds.RejectArguments{
+			Space:  req.Space,
+			Digest: req.Digest,
+		},
+		options...,
+	)
+	if err != nil {
+		return nil, nil, fmt.Errorf("creating reject invocation: %w", err)
+	}
+
+	return inv, prfs, nil
+}
+
 // ReplicaAllocateRequest contains the parameters for a /blob/replica/allocate invocation.
 type ReplicaAllocateRequest struct {
 	Space  did.DID
-	Digest []byte
+	Digest multihash.Multihash
 	Size   uint64
 	Site   ucan.Invocation // Location commitment
 	Cause  cid.Cid

pkg/service/handlers/blob_abort.go

@@ -0,0 +1,103 @@
+package handlers
+
+import (
+	"fmt"
+
+	blobcmds "github.com/fil-forge/libforge/commands/blob"
+	"github.com/fil-forge/libforge/digestutil"
+	ucanlib "github.com/fil-forge/libforge/ucan"
+	"github.com/fil-forge/sprue/pkg/piriclient"
+	"github.com/fil-forge/sprue/pkg/routing"
+	"github.com/fil-forge/sprue/pkg/store/agent"
+	"github.com/fil-forge/ucantone/binding"
+	"github.com/fil-forge/ucantone/errors"
+	"github.com/fil-forge/ucantone/server"
+	"github.com/fil-forge/ucantone/ucan"
+	"go.uber.org/zap"
+)
+
+// NewBlobAbortHandler abandons a space's in-flight upload of a parked
+// (never-accepted) blob: it recovers the storage node holding it from the
+// Cause receipt chain and forwards a /blob/reject there. Nothing is
+// deregistered — registration happens only at accept, which a parked blob
+// never reached.
+//
+// A cause that does not resolve to a known /blob/add task fails with
+// the named error MissingCause; a node refusing the translated reject
+// because the space has accepted the blob surfaces the node's BlobAccepted
+// as a named failure, so the client can distinguish "use /blob/remove"
+// from a retryable fault. Other forward errors are propagated as generic
+// receipt failures: abort mutates no local state, so the caller can simply
+// retry.
+func NewBlobAbortHandler(router *routing.Service, nodeProvider piriclient.Provider, agentStore agent.Store, logger *zap.Logger) server.Route {
+	log := logger.With(zap.Stringer("handler", blobcmds.Abort))
+	return blobcmds.Abort.Route(
+		func(req *binding.Request[*blobcmds.AbortArguments], res *binding.Response[*blobcmds.AbortOK]) error {
+			args := req.Task().Arguments()
+			space := req.Invocation().Subject()
+			log := log.With(
+				zap.Stringer("space", space),
+				zap.String("blob", digestutil.Format(args.Digest)),
+			)
+			log.Debug("aborting blob upload")
+
+			if !args.Cause.Defined() {
+				return res.SetFailure(blobcmds.ErrMissingCause)
+			}
+
+			provider, err := primaryProviderForBlob(req.Context(), agentStore, args.Cause)
+			if err != nil {
+				// An unknown cause — one whose receipt chain we don't hold —
+				// cannot route to a node; per the RFC it is the named error
+				// MissingCause, not a retryable execution fault.
+				if errors.Is(err, agent.ErrReceiptNotFound) || errors.Is(err, agent.ErrInvocationNotFound) {
+					log.Debug("cause does not resolve to a known blob add task", zap.Error(err))
+					return res.SetFailure(errors.New(blobcmds.MissingCauseErrorName,
+						"cause does not resolve to a known /blob/add task"))
+				}
+				log.Error("failed to recover provider from receipt chain", zap.Error(err))
+				return fmt.Errorf("recovering provider for parked blob: %w", err)
+			}
+
+			info, err := router.GetProviderInfo(req.Context(), provider)
+			if err != nil {
+				log.Error("failed to get provider info", zap.Error(err))
+				return fmt.Errorf("getting provider info: %w", err)
+			}
+			client, err := nodeProvider.Client(info.ID, info.Endpoint)
+			if err != nil {
+				log.Error("failed to create piri client", zap.Error(err))
+				return fmt.Errorf("creating piri client: %w", err)
+			}
+
+			// The proof chain for /blob/reject comes from the proofs the
+			// provider granted the upload service at registration.
+			proofStore := ucanlib.NewContainerProofStore(info.Proofs)
+			_, inv, rcpt, err := client.Reject(req.Context(), &piriclient.RejectRequest{
+				Space:  space,
+				Digest: args.Digest,
+			}, proofStore)
+			if err != nil {
+				// The node refuses to reject a blob this space has accepted.
+				// Surface the named failure rather than a generic fault so
+				// the client knows to use /blob/remove instead of retrying.
+				var named errors.Named
+				if errors.As(err, &named) && named.Name() == blobcmds.BlobAcceptedErrorName {
+					log.Debug("provider refused reject: blob accepted by space",
+						zap.Stringer("provider", provider))
+					return res.SetFailure(named)
+				}
+				log.Error("failed to execute reject on provider",
+					zap.Stringer("provider", provider), zap.Error(err))
+				return fmt.Errorf("executing reject on provider: %w", err)
+			}
+
+			if err := writeAgentMessage(req.Context(), agentStore, []ucan.Invocation{inv}, []ucan.Receipt{rcpt}); err != nil {
+				log.Error("failed to write agent message", zap.Error(err))
+				return fmt.Errorf("writing agent message: %w", err)
+			}
+
+			return res.SetSuccess(&blobcmds.AbortOK{})
+		},
+	)
+}