feat: blob removal — serve /blob/remove, /upload/remove, /blob/abort (FIL-522) #33
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -196,6 +196,142 @@ 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 []byte | ||
| } | ||
|
|
||
| // 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 []byte | ||
|
Member
There was a problem hiding this comment. Also here?
Member
Author
There was a problem hiding this comment. Done in 57c38ef (see thread above — applied to all the request structs in this file for consistency). |
||
| } | ||
|
|
||
| // 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 | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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{}) | ||
| }, | ||
| ) | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Shouldn't we use the expected type for this -
multihash.Multihash?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yep, good call — done in 57c38ef. The pre-existing request structs in this file (
AllocateRequest,AcceptRequest,ReplicaAllocateRequest) had the same[]bytepattern, so I converted all five tomultihash.Multihashrather than just the two new ones (plus theforwardBlobReleasehelper param). Assignment-compatible both ways, so no caller changes.