From 2e8f624a46606520f6de52d9c6a341a270bdde97 Mon Sep 17 00:00:00 2001 From: Joe Freeman Date: Tue, 8 Sep 2026 21:09:38 +0100 Subject: [PATCH 1/2] Support partial blob reads --- adapters/python/coflux/context.py | 17 +++- adapters/python/coflux/models.py | 16 ++++ adapters/python/coflux/protocol.py | 28 +++--- cli/internal/adapter/protocol.go | 13 ++- cli/internal/blob/blob.go | 78 ++++++++++++++++ cli/internal/blob/blob_test.go | 143 +++++++++++++++++++++++++++++ cli/internal/blob/http.go | 65 +++++++++++++ cli/internal/blob/s3.go | 45 +++++++++ cli/internal/pool/pool.go | 8 +- cli/internal/worker/worker.go | 27 ++++-- docs/docs/assets.md | 16 +++- docs/docs/blobs.md | 6 ++ tests/test_assets.py | 41 ++++++++- tests/test_epochs.py | 2 +- 14 files changed, 471 insertions(+), 34 deletions(-) create mode 100644 cli/internal/blob/blob_test.go diff --git a/adapters/python/coflux/context.py b/adapters/python/coflux/context.py index 3ddf8e3b..fceb1399 100644 --- a/adapters/python/coflux/context.py +++ b/adapters/python/coflux/context.py @@ -338,12 +338,21 @@ def get_asset_entries(self, asset_id: str) -> list[AssetEntry]: entries.append(AssetEntry(path, blob_key, size, metadata or {})) return entries - def download_blob(self, blob_key: str, target_path: Path) -> None: - """Download a blob to a local file.""" + def download_blob( + self, + blob_key: str, + target_path: Path, + *, + offset: int | None = None, + length: int | None = None, + ) -> None: + """Download a blob, or a byte range of one, to a local file.""" request_id = protocol.request_download_blob( self.execution_id, blob_key, str(target_path), + offset, + length, ) self._wait_response(request_id) @@ -450,10 +459,10 @@ def create_asset( if not paths_to_upload and not resolved_entries: raise ValueError("No files found to create asset") - abs_paths = [str(p) for _, p in paths_to_upload] if paths_to_upload else None + upload_paths = {rel: str(p) for rel, p in paths_to_upload} or None request_id = protocol.request_persist_asset( self.execution_id, - abs_paths, + upload_paths, {"name": name} if name else None, resolved_entries if resolved_entries else None, ) diff --git a/adapters/python/coflux/models.py b/adapters/python/coflux/models.py index 3991db4f..ec8c9f6e 100644 --- a/adapters/python/coflux/models.py +++ b/adapters/python/coflux/models.py @@ -4,6 +4,7 @@ import fnmatch import functools +import tempfile import typing as t from pathlib import Path @@ -42,6 +43,21 @@ def restore(self, *, at: Path | str | None = None) -> Path: ctx.download_blob(self.blob_key, target) return target + def read(self, offset: int = 0, length: int | None = None) -> bytes: + """Read bytes from this entry without restoring the whole file. + + Reads to the end of the file when ``length`` is omitted. + + Useful for formats that seek rather than read straight through — a + Parquet footer, say — where restoring a large file to read a few + kilobytes of it would be wasteful. + """ + ctx = get_context() + with tempfile.TemporaryDirectory() as directory: + target = Path(directory) / "range" + ctx.download_blob(self.blob_key, target, offset=offset, length=length) + return target.read_bytes() + class AssetMetadata(t.NamedTuple): """Metadata for an asset reference.""" diff --git a/adapters/python/coflux/protocol.py b/adapters/python/coflux/protocol.py index 23932b38..f1f57689 100644 --- a/adapters/python/coflux/protocol.py +++ b/adapters/python/coflux/protocol.py @@ -265,7 +265,7 @@ def request_select( def request_persist_asset( execution_id: str, - paths: list[str] | None = None, + paths: dict[str, str] | None = None, metadata: dict[str, Any] | None = None, entries: dict[str, tuple[str, int, dict[str, Any]]] | None = None, ) -> int: @@ -273,7 +273,7 @@ def request_persist_asset( Args: execution_id: The execution this asset belongs to. - paths: Local file paths to upload and include. + paths: Local files to upload, as {path within the asset: local path}. metadata: Asset-level metadata (e.g. name). entries: Pre-resolved entries referencing existing blobs. Each value is (blob_key, size, entry_metadata). @@ -311,16 +311,22 @@ def request_download_blob( execution_id: str, blob_key: str, target_path: str, + offset: int | None = None, + length: int | None = None, ) -> int: - """Request to download a blob to a local file.""" - return get_protocol().send_request( - "download_blob", - { - "execution_id": execution_id, - "blob_key": blob_key, - "target_path": target_path, - }, - ) + """Request to download a blob, or a byte range of one, to a local file.""" + params: dict[str, Any] = { + "execution_id": execution_id, + "blob_key": blob_key, + "target_path": target_path, + } + # Omitted rather than sent as null, so the message keeps the shape + # older CLIs expect. + if offset is not None: + params["offset"] = offset + if length is not None: + params["length"] = length + return get_protocol().send_request("download_blob", params) def request_upload_blob( diff --git a/cli/internal/adapter/protocol.go b/cli/internal/adapter/protocol.go index 126cccdb..c3c32a97 100644 --- a/cli/internal/adapter/protocol.go +++ b/cli/internal/adapter/protocol.go @@ -234,10 +234,10 @@ type SelectParams struct { // PersistAssetParams for persist_asset request type PersistAssetParams struct { - ExecutionID string `json:"execution_id"` - Paths []string `json:"paths,omitempty"` - Metadata map[string]any `json:"metadata,omitempty"` - Entries map[string][]any `json:"entries,omitempty"` // Pre-resolved entries: {path: [blob_key, size, metadata]} + ExecutionID string `json:"execution_id"` + Paths map[string]string `json:"paths,omitempty"` // Files to upload: {path within the asset: local path} + Metadata map[string]any `json:"metadata,omitempty"` + Entries map[string][]any `json:"entries,omitempty"` // Pre-resolved entries: {path: [blob_key, size, metadata]} } // PersistAssetResult is the response to persist_asset @@ -434,11 +434,14 @@ type StreamClosedParams struct { Error map[string]any `json:"error,omitempty"` } -// DownloadBlobParams for download_blob request +// DownloadBlobParams for download_blob request. Offset and length are +// optional: with neither, the whole blob is downloaded. type DownloadBlobParams struct { ExecutionID string `json:"execution_id"` BlobKey string `json:"blob_key"` TargetPath string `json:"target_path"` + Offset *int64 `json:"offset,omitempty"` + Length *int64 `json:"length,omitempty"` } // UploadBlobParams for upload_blob request diff --git a/cli/internal/blob/blob.go b/cli/internal/blob/blob.go index fbadeebe..e5b3dcb9 100644 --- a/cli/internal/blob/blob.go +++ b/cli/internal/blob/blob.go @@ -14,6 +14,11 @@ import ( type Store interface { // Get retrieves a blob by key, returns nil if not found Get(key string) (io.ReadCloser, error) + // GetRange retrieves a byte range of a blob, returns nil if not found. + // A negative length reads to the end of the blob. + GetRange(key string, offset, length int64) (io.ReadCloser, error) + // Exists reports whether a blob is already stored + Exists(key string) (bool, error) // Put stores a blob and returns its key (content-addressed) Put(reader io.Reader) (string, error) // Upload uploads a file and returns its key @@ -22,6 +27,34 @@ type Store interface { Download(key, path string) (bool, error) } +// Blobs at or above this size are checked for existence before being +// uploaded. +const existsCheckThreshold = 1 << 20 // 1 MiB + +// skipUpload reports whether content with this key is already stored, and +// so needn't be sent again. Content addressing makes re-uploading it +// redundant, but asking costs a round trip, so it's only worth it once the +// content is big enough that re-sending would cost more than asking. +// +// A failed check reports "not stored": this is only an optimisation, and +// uploading something that's already there is always safe. +func skipUpload(store Store, key string, size int) bool { + if size < existsCheckThreshold { + return false + } + exists, err := store.Exists(key) + return err == nil && exists +} + +// rangeHeader formats an HTTP byte range. A negative length means "to the +// end of the blob". +func rangeHeader(offset, length int64) string { + if length < 0 { + return fmt.Sprintf("bytes=%d-", offset) + } + return fmt.Sprintf("bytes=%d-%d", offset, offset+length-1) +} + // Manager manages multiple blob stores with fallback type Manager struct { stores []Store @@ -52,6 +85,24 @@ func (m *Manager) Get(key string) (io.ReadCloser, error) { return nil, fmt.Errorf("blob not found: %s", key) } +// GetRange retrieves a byte range of a blob from any store. A negative +// length reads to the end of the blob. +func (m *Manager) GetRange(key string, offset, length int64) (io.ReadCloser, error) { + // Deliberately not served from the cache: it holds whole blobs, and + // Download treats the file merely existing as a complete one, so a + // range must never be written there. + for _, store := range m.stores { + reader, err := store.GetRange(key, offset, length) + if err != nil { + return nil, err + } + if reader != nil { + return reader, nil + } + } + return nil, fmt.Errorf("blob not found: %s", key) +} + // Put stores a blob in the first store func (m *Manager) Put(reader io.Reader) (string, error) { if len(m.stores) == 0 { @@ -119,6 +170,33 @@ func (m *Manager) DownloadTo(key, targetPath string) error { return fmt.Errorf("blob not found: %s", key) } +// DownloadRangeTo downloads a byte range of a blob to a specific path. A +// negative length reads to the end of the blob. +func (m *Manager) DownloadRangeTo(key, targetPath string, offset, length int64) error { + if err := os.MkdirAll(filepath.Dir(targetPath), 0755); err != nil { + return err + } + + f, err := os.Create(targetPath) + if err != nil { + return err + } + defer func() { _ = f.Close() }() + + if length == 0 { + return nil + } + + reader, err := m.GetRange(key, offset, length) + if err != nil { + return err + } + defer func() { _ = reader.Close() }() + + _, err = io.Copy(f, reader) + return err +} + // CachePath returns the cache path for a blob key func (m *Manager) CachePath(key string) string { // Use first 4 chars for directory sharding diff --git a/cli/internal/blob/blob_test.go b/cli/internal/blob/blob_test.go new file mode 100644 index 00000000..044455f0 --- /dev/null +++ b/cli/internal/blob/blob_test.go @@ -0,0 +1,143 @@ +package blob + +import ( + "bytes" + "errors" + "fmt" + "io" + "testing" +) + +// fakeStore records what it was asked, so a test can tell an upload that +// happened from one that was skipped. +type fakeStore struct { + content map[string][]byte + existsErr error + existsCalls int +} + +func (f *fakeStore) Get(key string) (io.ReadCloser, error) { + content, ok := f.content[key] + if !ok { + return nil, nil + } + return io.NopCloser(bytes.NewReader(content)), nil +} + +func (f *fakeStore) GetRange(key string, offset, length int64) (io.ReadCloser, error) { + content, ok := f.content[key] + if !ok { + return nil, nil + } + if offset > int64(len(content)) { + return nil, fmt.Errorf("range not satisfiable") + } + end := int64(len(content)) + if length >= 0 && offset+length < end { + end = offset + length + } + return io.NopCloser(bytes.NewReader(content[offset:end])), nil +} + +func (f *fakeStore) Exists(key string) (bool, error) { + f.existsCalls++ + if f.existsErr != nil { + return false, f.existsErr + } + _, ok := f.content[key] + return ok, nil +} + +func (f *fakeStore) Put(reader io.Reader) (string, error) { return "", nil } +func (f *fakeStore) Upload(path string) (string, error) { return "", nil } +func (f *fakeStore) Download(key, path string) (bool, error) { return false, nil } + +func stored(key string, content string) *fakeStore { + return &fakeStore{content: map[string][]byte{key: []byte(content)}} +} + +func TestSkipUploadBelowThreshold(t *testing.T) { + store := stored("abc", "hello") + if skipUpload(store, "abc", existsCheckThreshold-1) { + t.Fatal("expected small content to be uploaded without checking") + } + if store.existsCalls != 0 { + t.Fatalf("expected no existence check, got %d", store.existsCalls) + } +} + +func TestSkipUploadWhenPresent(t *testing.T) { + if !skipUpload(stored("abc", "hello"), "abc", existsCheckThreshold) { + t.Fatal("expected stored content to be skipped") + } +} + +func TestSkipUploadWhenAbsent(t *testing.T) { + if skipUpload(stored("abc", "hello"), "other", existsCheckThreshold) { + t.Fatal("expected absent content to be uploaded") + } +} + +func TestSkipUploadOnCheckFailure(t *testing.T) { + // The check is an optimisation, so a store that can't answer must not + // stop the upload. + store := stored("abc", "hello") + store.existsErr = errors.New("nope") + if skipUpload(store, "abc", existsCheckThreshold) { + t.Fatal("expected a failed check to fall back to uploading") + } +} + +func TestRangeHeader(t *testing.T) { + for _, c := range []struct { + offset, length int64 + want string + }{ + {0, 16, "bytes=0-15"}, + {1000, 16, "bytes=1000-1015"}, + {1000, -1, "bytes=1000-"}, + } { + if got := rangeHeader(c.offset, c.length); got != c.want { + t.Errorf("rangeHeader(%d, %d) = %q, want %q", c.offset, c.length, got, c.want) + } + } +} + +func TestManagerGetRangeFallsThroughStores(t *testing.T) { + // The blob is missing from the first store, so the second answers. + m := NewManager([]Store{stored("other", "xxx"), stored("abc", "0123456789")}, t.TempDir(), 200) + + reader, err := m.GetRange("abc", 3, 4) + if err != nil { + t.Fatal(err) + } + defer func() { _ = reader.Close() }() + got, err := io.ReadAll(reader) + if err != nil { + t.Fatal(err) + } + if string(got) != "3456" { + t.Fatalf("got %q, want %q", got, "3456") + } +} + +func TestManagerGetRangeToEnd(t *testing.T) { + m := NewManager([]Store{stored("abc", "0123456789")}, t.TempDir(), 200) + + reader, err := m.GetRange("abc", 6, -1) + if err != nil { + t.Fatal(err) + } + defer func() { _ = reader.Close() }() + got, _ := io.ReadAll(reader) + if string(got) != "6789" { + t.Fatalf("got %q, want %q", got, "6789") + } +} + +func TestManagerGetRangeNotFound(t *testing.T) { + m := NewManager([]Store{stored("abc", "0123456789")}, t.TempDir(), 200) + if _, err := m.GetRange("missing", 0, 4); err == nil { + t.Fatal("expected an error for a blob no store holds") + } +} diff --git a/cli/internal/blob/http.go b/cli/internal/blob/http.go index 3a683cca..435a7751 100644 --- a/cli/internal/blob/http.go +++ b/cli/internal/blob/http.go @@ -55,6 +55,67 @@ func (s *HTTPStore) Get(key string) (io.ReadCloser, error) { return resp.Body, nil } +// GetRange retrieves a byte range of a blob +func (s *HTTPStore) GetRange(key string, offset, length int64) (io.ReadCloser, error) { + if length == 0 { + return io.NopCloser(bytes.NewReader(nil)), nil + } + url := fmt.Sprintf("%s/%s", s.baseURL, key) + req, err := http.NewRequest(http.MethodGet, url, nil) + if err != nil { + return nil, err + } + req.Header.Set("Range", rangeHeader(offset, length)) + if s.token != "" { + req.Header.Set("Authorization", "Bearer "+s.token) + } + if s.project != "" { + req.Header.Set("X-Project", s.project) + } + resp, err := s.client.Do(req) + if err != nil { + return nil, err + } + if resp.StatusCode == http.StatusNotFound { + _ = resp.Body.Close() + return nil, nil + } + // A store that ignores the range header answers 200 with the whole + // blob, which would silently be the wrong bytes. + if resp.StatusCode != http.StatusPartialContent { + _ = resp.Body.Close() + return nil, fmt.Errorf("HTTP %d: %s", resp.StatusCode, resp.Status) + } + return resp.Body, nil +} + +// Exists reports whether a blob is already stored +func (s *HTTPStore) Exists(key string) (bool, error) { + url := fmt.Sprintf("%s/%s", s.baseURL, key) + req, err := http.NewRequest(http.MethodHead, url, nil) + if err != nil { + return false, err + } + if s.token != "" { + req.Header.Set("Authorization", "Bearer "+s.token) + } + if s.project != "" { + req.Header.Set("X-Project", s.project) + } + resp, err := s.client.Do(req) + if err != nil { + return false, err + } + _ = resp.Body.Close() + if resp.StatusCode == http.StatusNotFound { + return false, nil + } + if resp.StatusCode != http.StatusOK { + return false, fmt.Errorf("HTTP %d: %s", resp.StatusCode, resp.Status) + } + return true, nil +} + // Put stores a blob and returns its key func (s *HTTPStore) Put(reader io.Reader) (string, error) { // Read content to compute key and store @@ -68,6 +129,10 @@ func (s *HTTPStore) Put(reader io.Reader) (string, error) { return "", err } + if skipUpload(s, key, len(content)) { + return key, nil + } + url := fmt.Sprintf("%s/%s", s.baseURL, key) req, err := http.NewRequest(http.MethodPut, url, bytes.NewReader(content)) if err != nil { diff --git a/cli/internal/blob/s3.go b/cli/internal/blob/s3.go index 55b1eac7..da590e53 100644 --- a/cli/internal/blob/s3.go +++ b/cli/internal/blob/s3.go @@ -80,6 +80,47 @@ func (s *S3Store) Get(key string) (io.ReadCloser, error) { return output.Body, nil } +// GetRange retrieves a byte range of a blob +func (s *S3Store) GetRange(key string, offset, length int64) (io.ReadCloser, error) { + if length == 0 { + return io.NopCloser(bytes.NewReader(nil)), nil + } + ctx := context.Background() + output, err := s.client.GetObject(ctx, &s3.GetObjectInput{ + Bucket: aws.String(s.bucket), + Key: aws.String(s.s3Key(key)), + Range: aws.String(rangeHeader(offset, length)), + }) + if err != nil { + // Check for not found errors + var noSuchKey *types.NoSuchKey + var notFound *types.NotFound + if errors.As(err, &noSuchKey) || errors.As(err, ¬Found) { + return nil, nil + } + return nil, fmt.Errorf("failed to get S3 object range: %w", err) + } + return output.Body, nil +} + +// Exists reports whether a blob is already stored +func (s *S3Store) Exists(key string) (bool, error) { + ctx := context.Background() + _, err := s.client.HeadObject(ctx, &s3.HeadObjectInput{ + Bucket: aws.String(s.bucket), + Key: aws.String(s.s3Key(key)), + }) + if err != nil { + var noSuchKey *types.NoSuchKey + var notFound *types.NotFound + if errors.As(err, &noSuchKey) || errors.As(err, ¬Found) { + return false, nil + } + return false, fmt.Errorf("failed to head S3 object: %w", err) + } + return true, nil +} + // Put stores a blob and returns its key func (s *S3Store) Put(reader io.Reader) (string, error) { // Read content to compute key @@ -93,6 +134,10 @@ func (s *S3Store) Put(reader io.Reader) (string, error) { return "", err } + if skipUpload(s, key, len(content)) { + return key, nil + } + ctx := context.Background() _, err = s.client.PutObject(ctx, &s3.PutObjectInput{ Bucket: aws.String(s.bucket), diff --git a/cli/internal/pool/pool.go b/cli/internal/pool/pool.go index f2fa9d7a..b65e53e2 100644 --- a/cli/internal/pool/pool.go +++ b/cli/internal/pool/pool.go @@ -19,11 +19,11 @@ type ExecutionHandler interface { // Select waits for the first of one or more handles (executions/inputs) to resolve Select(ctx context.Context, params *adapter.SelectParams) (*adapter.SelectResult, error) // PersistAsset persists files as an asset - PersistAsset(ctx context.Context, executionID string, paths []string, metadata map[string]any, preResolved map[string][]any) (map[string]any, error) + PersistAsset(ctx context.Context, executionID string, paths map[string]string, metadata map[string]any, preResolved map[string][]any) (map[string]any, error) // GetAsset retrieves asset entries GetAsset(ctx context.Context, executionID string, assetID string) (map[string]any, error) - // DownloadBlob downloads a blob to a local file - DownloadBlob(ctx context.Context, executionID, blobKey, targetPath string) error + // DownloadBlob downloads a blob, or a byte range of one, to a local file + DownloadBlob(ctx context.Context, executionID, blobKey, targetPath string, offset, length *int64) error // UploadBlob uploads a local file as a blob UploadBlob(ctx context.Context, executionID, sourcePath string) (string, error) // Suspend suspends an execution @@ -793,7 +793,7 @@ func (p *Pool) handleRequest(ctx context.Context, exec *adapter.Executor, method errInfo = &adapter.ErrorInfo{Code: "parse_error", Message: err.Error()} break } - if err := p.handler.DownloadBlob(ctx, req.ExecutionID, req.BlobKey, req.TargetPath); err != nil { + if err := p.handler.DownloadBlob(ctx, req.ExecutionID, req.BlobKey, req.TargetPath, req.Offset, req.Length); err != nil { errInfo = &adapter.ErrorInfo{Code: "download_error", Message: err.Error()} } else { result = map[string]any{} diff --git a/cli/internal/worker/worker.go b/cli/internal/worker/worker.go index 4cb3f2a3..67019653 100644 --- a/cli/internal/worker/worker.go +++ b/cli/internal/worker/worker.go @@ -1272,7 +1272,7 @@ func (w *Worker) SubmitInput(ctx context.Context, params *adapter.SubmitInputPar return inputExternalID, nil } -func (w *Worker) PersistAsset(ctx context.Context, executionID string, paths []string, metadata map[string]any, preResolved map[string][]any) (map[string]any, error) { +func (w *Worker) PersistAsset(ctx context.Context, executionID string, paths map[string]string, metadata map[string]any, preResolved map[string][]any) (map[string]any, error) { // Upload each file and create entries // Server format: {path: [blob_key, size, metadata]} entries := make(map[string][]any) @@ -1282,8 +1282,10 @@ func (w *Worker) PersistAsset(ctx context.Context, executionID string, paths []s entries[path] = entry } - // Upload local files - for _, path := range paths { + // Upload local files. The key is the path within the asset, which is + // not the file's basename: an asset can hold a directory tree, and two + // files in it can share a name. + for assetPath, path := range paths { key, err := w.blobs.Upload(path) if err != nil { return nil, fmt.Errorf("failed to upload %s: %w", path, err) @@ -1300,7 +1302,7 @@ func (w *Worker) PersistAsset(ctx context.Context, executionID string, paths []s entryMetadata["type"] = mimeType } } - entries[filepath.Base(path)] = []any{key, size, entryMetadata} + entries[assetPath] = []any{key, size, entryMetadata} } // Get asset name from metadata if provided @@ -1446,9 +1448,22 @@ func (s checkpointSink) SetCheckpoints(ctx context.Context, executionID string, return err } -func (w *Worker) DownloadBlob(ctx context.Context, executionID, blobKey, targetPath string) error { +func (w *Worker) DownloadBlob(ctx context.Context, executionID, blobKey, targetPath string, offset, length *int64) error { // Download blob to the target path - return w.blobs.DownloadTo(blobKey, targetPath) + if offset == nil && length == nil { + return w.blobs.DownloadTo(blobKey, targetPath) + } + var start int64 + if offset != nil { + start = *offset + } + // A negative length reads to the end of the blob, which is what an + // offset with no length means. + size := int64(-1) + if length != nil { + size = *length + } + return w.blobs.DownloadRangeTo(blobKey, targetPath, start, size) } func (w *Worker) UploadBlob(ctx context.Context, executionID, sourcePath string) (string, error) { diff --git a/docs/docs/assets.md b/docs/docs/assets.md index 443d5458..e4fccda6 100644 --- a/docs/docs/assets.md +++ b/docs/docs/assets.md @@ -59,5 +59,19 @@ path.read_text() By default an asset is restored to the task's temporary directory, at the same relative path that it was persisted from. To change this, the `at` argument can be specified (as a `pathlib.Path`, or string): ```python -asset.restore(to="other/dir") +asset.restore(at="other/dir") ``` + +## Reading part of an entry + +An entry can be read without restoring the whole file, by giving an offset and a length in bytes: + +```python +entry = asset["data.parquet"] +footer = entry.read(entry.size - 8) # to the end of the file +header = entry.read(0, 4) # the first four bytes +``` + +`read()` with no arguments returns the whole entry. Omitting the length reads to the end. + +This is for formats that seek rather than read straight through — a Parquet footer, say, where restoring a large file to read a few kilobytes of it would be wasteful. For anything you're going to read in full, `restore()` is simpler and puts the file on disk where other tools can reach it. diff --git a/docs/docs/blobs.md b/docs/docs/blobs.md index 62845341..b58cf52d 100644 --- a/docs/docs/blobs.md +++ b/docs/docs/blobs.md @@ -26,6 +26,12 @@ threshold = 100 To have all values stored as blobs, set the threshold to zero. +## Deduplication + +Blobs are content-addressed — a blob's key is the SHA-256 of its contents — so storing the same data twice is redundant. Above a size threshold, the store is asked whether it already holds the content, and the upload is skipped if it does. Below that threshold the check isn't worth its round trip, and the data is simply written again. + +This matters when an asset is rebuilt from a previous one: the files that haven't changed cost nothing to persist again, in storage or in transfer. + ## S3 blob store As an alternative to the built-in blob store, AWS S3 can be used. To enable this, update the configuration file: diff --git a/tests/test_assets.py b/tests/test_assets.py index 3ca9d95e..9ef64fca 100644 --- a/tests/test_assets.py +++ b/tests/test_assets.py @@ -31,7 +31,7 @@ def test_persist_and_get_asset(worker, tmp_path): # Persist the file as an asset asset_result = ex1.conn.persist_asset( ex1.execution_id, - [str(asset_file)], + {"asset.txt": str(asset_file)}, metadata={"name": "my_asset"}, ) assert "asset_id" in asset_result @@ -89,7 +89,7 @@ def test_asset_inspect_and_download(worker, tmp_path): asset_result = ex1.conn.persist_asset( ex1.execution_id, - [src_path], + {"data.txt": src_path}, metadata={"name": "cli_test_asset"}, ) asset_id = asset_result["asset_id"] @@ -126,3 +126,40 @@ def test_asset_inspect_and_download(worker, tmp_path): ctx.get_blob(blob_key, blob_output) with open(blob_output) as f: assert f.read() == "asset content for CLI test" + + +def test_asset_entry_paths_are_not_flattened(worker, tmp_path): + """An asset holds a directory tree, so entries keep their own paths. + + Two files can share a basename in different directories, which is why + the path within the asset is chosen by the caller rather than derived + from the file being uploaded. + """ + targets = [workflow("test", "main"), task("test", "producer")] + + with worker(targets, concurrency=2) as ctx: + ctx.submit("test", "main") + + ex0 = ctx.executor.next_execute() + ref_prod = ex0.conn.submit_task(ex0.execution_id, "test", "producer", []) + ex1 = ctx.executor.next_execute() + + first = tmp_path / "first.txt" + first.write_text("one") + second = tmp_path / "second.txt" + second.write_text("two") + + asset_result = ex1.conn.persist_asset( + ex1.execution_id, + {"a/same.txt": str(first), "a/b/same.txt": str(second)}, + metadata={"name": "tree"}, + ) + asset_id = asset_result["asset_id"] + + ex1.conn.complete(ex1.execution_id, value="produced") + assert ex0.conn.resolve(ex0.execution_id, ref_prod)["value"] == "produced" + + entries = ex0.conn.get_asset(ex0.execution_id, asset_id)["entries"] + assert sorted(entries) == ["a/b/same.txt", "a/same.txt"] + # Distinct content, so distinct blobs — neither overwrote the other. + assert entries["a/same.txt"][0] != entries["a/b/same.txt"][0] diff --git a/tests/test_epochs.py b/tests/test_epochs.py index e4d443ce..cf857b1b 100644 --- a/tests/test_epochs.py +++ b/tests/test_epochs.py @@ -351,7 +351,7 @@ def test_asset_reference_across_epoch_boundary(isolated_server, tmp_path): asset_result = ex1.conn.persist_asset( ex1.execution_id, - [asset_file], + {"epoch_asset.txt": asset_file}, metadata={"name": "epoch_asset"}, ) assert "asset_id" in asset_result From 6f9e288745f1cb2d138e68d87292119f5bd43a37 Mon Sep 17 00:00:00 2001 From: Joe Freeman Date: Tue, 8 Sep 2026 21:31:40 +0100 Subject: [PATCH 2/2] Close Sqlite file before moving --- server/lib/coflux/orchestration/server.ex | 25 +++++++++++++++++++--- server/lib/coflux/store/epochs.ex | 26 ++++++++++++++++------- 2 files changed, 40 insertions(+), 11 deletions(-) diff --git a/server/lib/coflux/orchestration/server.ex b/server/lib/coflux/orchestration/server.ex index 6f6836ca..53877fe2 100644 --- a/server/lib/coflux/orchestration/server.ex +++ b/server/lib/coflux/orchestration/server.ex @@ -6137,13 +6137,32 @@ defmodule Coflux.Orchestration.Server do # Searches archived epochs across both tiers (unindexed, then indexed via Bloom). # `query_fn` receives an archive DB handle and returns `{:found, result}` or `:not_found`. # `bloom_fn` receives the epoch index and returns candidate epoch IDs. + # Query one archived epoch, treating an unreadable file as a miss. + # + # A corrupt archive opens cleanly — SQLite only reads the schema when a + # statement is prepared — so the failure lands inside the query, where + # `Store` matches on success. Without this, one bad file takes the + # project's orchestration server down on every lookup that reaches the + # archives, rather than costing that lookup one epoch's worth of rows. + defp query_epoch(state, epoch_id, archive_db, query_fn) do + query_fn.(archive_db) + rescue + error -> + Logger.error( + "Couldn't read archived epoch #{epoch_id} in project #{state.project_id}: " <> + Exception.message(error) + ) + + :not_found + end + defp search_archived_epochs(state, query_fn, bloom_fn) do # Tier 1: Check unindexed DBs (always open, newest first) unindexed = Epochs.unindexed_dbs(state.epochs) result = - Enum.reduce_while(unindexed, :not_found, fn {_epoch_id, archive_db}, :not_found -> - case query_fn.(archive_db) do + Enum.reduce_while(unindexed, :not_found, fn {epoch_id, archive_db}, :not_found -> + case query_epoch(state, epoch_id, archive_db, query_fn) do {:found, _} = found -> {:halt, found} :not_found -> {:cont, :not_found} end @@ -6167,7 +6186,7 @@ defmodule Coflux.Orchestration.Server do case Exqlite.Sqlite3.open(path) do {:ok, archive_db} -> try do - case query_fn.(archive_db) do + case query_epoch(state, epoch_id, archive_db, query_fn) do {:found, _} = found -> {:halt, found} :not_found -> {:cont, :not_found} end diff --git a/server/lib/coflux/store/epochs.ex b/server/lib/coflux/store/epochs.ex index dce54dbf..955fcfc1 100644 --- a/server/lib/coflux/store/epochs.ex +++ b/server/lib/coflux/store/epochs.ex @@ -85,32 +85,42 @@ defmodule Coflux.Store.Epochs do function, so that the index always knows about all archived epochs. The current active file is renamed to the archive path and a fresh - active file is created. The old DB handle remains valid (Linux fd - semantics) and moves to the unindexed list. + active file is created. The archived database is reopened at its new + path and moves to the unindexed list. - Returns {:ok, new_epoch_state, old_db}. + Returns {:ok, new_epoch_state, archived_db}. """ def rotate(%__MODULE__{} = state, epoch_id) do active = active_path(state.project_id, state.name) archive = Path.join(archive_dir(state.project_id, state.name), "#{epoch_id}.sqlite") - # Rename current active → archive (old fd remains valid) + # Close before renaming, and reopen at the path the file now has. + # + # A handle carried across the rename keeps the fd, but SQLite derives + # the journal path from the path the connection was *opened* with — so + # it would go on journalling to `active`, which by then holds the next + # epoch's database. The first read transaction on such a handle finds + # the successor's journal sitting at its own journal path, takes it for + # a hot journal, rolls those pages into the file its fd points at, and + # truncates that file to the size the journal header records. The + # archive ends up with its successor's page 1, truncated to its + # successor's length: "malformed database schema - invalid rootpage". + :ok = Sqlite3.close(state.active_db) :ok = File.rename(active, archive) + {:ok, archived_db} = Sqlite3.open(archive) # Create fresh active file {:ok, new_db} = Sqlite3.open(active) :ok = Migrations.run(new_db, state.name) - old_db = state.active_db - new_state = %{ state | active_db: new_db, - unindexed: state.unindexed ++ [{epoch_id, old_db}], + unindexed: state.unindexed ++ [{epoch_id, archived_db}], archived_ids: state.archived_ids ++ [epoch_id] } - {:ok, new_state, old_db} + {:ok, new_state, archived_db} end @doc """