feat(coordinator): artifact upload/download endpoints (CTX-05, part 2)
Wire the artifact storage foundation to HTTP.
- PUT /tasks/{id}/artifacts/{filename}: a worker streams a partial result;
the coordinator verifies lease ownership (foreign worker → 409), streams
the bytes to blob storage while hashing, and records the metadata. An
orphaned blob from a failed metadata insert is cleaned up.
- GET /artifacts/{id}/download: streams an artifact back with its content
type, length, and checksum.
- Ownership is read with a new non-locking TaskRepository.Get, so no row lock
is held across a long upload. Identity travels in X-Worker-ID / X-Task-Attempt
headers per the contract; upload/download bypass the short request timeout.
- docker-compose mounts ./data for durable artifact storage; smoke and
requests.http exercise an upload → foreign-409 → download round-trip.
This commit is contained in:
@@ -0,0 +1,80 @@
|
||||
package usecase
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
|
||||
)
|
||||
|
||||
// UploadArtifact stores a worker's partial-result bytes and records the metadata.
|
||||
type UploadArtifact struct {
|
||||
tasks TaskRepository
|
||||
artifacts ArtifactRepository
|
||||
blobs BlobStore
|
||||
clk Clock
|
||||
}
|
||||
|
||||
func NewUploadArtifact(tasks TaskRepository, artifacts ArtifactRepository,
|
||||
blobs BlobStore, clk Clock) *UploadArtifact {
|
||||
return &UploadArtifact{tasks: tasks, artifacts: artifacts, blobs: blobs, clk: clk}
|
||||
}
|
||||
|
||||
func (uc *UploadArtifact) Execute(ctx context.Context, in UploadArtifactInput) (*domain.Artifact, error) {
|
||||
task, err := uc.tasks.Get(ctx, in.TaskID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Only the worker holding the current lease at this attempt may upload the
|
||||
// task's output — the coordinator never trusts an ownership claim on faith.
|
||||
if !task.IsLeaseHeldBy(in.WorkerID, in.Attempt) {
|
||||
return nil, domain.ErrLeaseConflict
|
||||
}
|
||||
|
||||
taskID := task.ID
|
||||
art, err := domain.NewArtifact(task.JobID, &taskID, domain.ArtifactPartialResult,
|
||||
in.Filename, in.ContentType, uc.clk.Now())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Stream to storage first: size and checksum are measured here, by us, not
|
||||
// taken from the worker. A large shard never sits in memory.
|
||||
sum, size, err := uc.blobs.Put(ctx, art.StorageKey, in.Body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
art.SetContent(sum, size)
|
||||
|
||||
// Persist the record. If that fails the blob would be an orphan, so remove it.
|
||||
if err := uc.artifacts.Insert(ctx, art); err != nil {
|
||||
_ = uc.blobs.Delete(ctx, art.StorageKey)
|
||||
return nil, err
|
||||
}
|
||||
return art, nil
|
||||
}
|
||||
|
||||
// DownloadArtifact returns an artifact's metadata together with a reader over
|
||||
// its bytes. The caller must close the reader.
|
||||
type DownloadArtifact struct {
|
||||
artifacts ArtifactRepository
|
||||
blobs BlobStore
|
||||
}
|
||||
|
||||
func NewDownloadArtifact(artifacts ArtifactRepository, blobs BlobStore) *DownloadArtifact {
|
||||
return &DownloadArtifact{artifacts: artifacts, blobs: blobs}
|
||||
}
|
||||
|
||||
func (uc *DownloadArtifact) Execute(ctx context.Context, id uuid.UUID) (*domain.Artifact, io.ReadCloser, error) {
|
||||
a, err := uc.artifacts.Get(ctx, id)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
rc, err := uc.blobs.Open(ctx, a.StorageKey)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return a, rc, nil
|
||||
}
|
||||
Reference in New Issue
Block a user