From d97ec4367c50c55d5bf70d14abbea09d0dd39507 Mon Sep 17 00:00:00 2001 From: Javier Rodriguez Date: Thu, 13 Aug 2026 12:14:07 +0200 Subject: [PATCH] feat(artifact-cas): stream raw uploads to the OCI CAS backend The OCI backend buffered the whole artifact in memory before pushing it to the registry, so CAS memory usage scaled with artifact size. This adds a streaming upload path that pushes the raw layer directly from the client stream, keeping CAS memory bounded regardless of artifact size. A custom v1.Layer streams the content to the registry with the known digest and size, detects the media type from a bounded peek, and verifies the content digest as the bytes stream. The artifact size is derived from the upload reader and carried to the backend so the layer can be built without buffering; uploads whose size is not known fall back to the buffered path. The download path is unchanged. Assisted-by: Claude Code Signed-off-by: Javier Rodriguez Chainloop-Trace-Sessions: 6b760aee-b968-4bf3-abb0-3f8ffd0d5143 --- .../internal/service/bytestream.go | 20 +-- pkg/blobmanager/backend.go | 4 +- pkg/blobmanager/oci/backend.go | 107 +++++++++++++-- pkg/blobmanager/oci/backend_streaming_test.go | 127 +++++++++++++++++ pkg/blobmanager/oci/backend_test.go | 21 +-- pkg/blobmanager/oci/stream_layer.go | 129 ++++++++++++++++++ pkg/blobmanager/oci/stream_layer_test.go | 121 ++++++++++++++++ pkg/casclient/uploader.go | 49 ++++++- pkg/casclient/uploader_test.go | 64 ++++++++- 9 files changed, 599 insertions(+), 43 deletions(-) create mode 100644 pkg/blobmanager/oci/backend_streaming_test.go create mode 100644 pkg/blobmanager/oci/stream_layer.go create mode 100644 pkg/blobmanager/oci/stream_layer_test.go diff --git a/app/artifact-cas/internal/service/bytestream.go b/app/artifact-cas/internal/service/bytestream.go index 86c588141..6edc5767d 100644 --- a/app/artifact-cas/internal/service/bytestream.go +++ b/app/artifact-cas/internal/service/bytestream.go @@ -116,11 +116,11 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro s.log.Infow("msg", "artifact does not exist, uploading", "digest", req.resource.Digest, "name", req.resource.FileName) - // Streaming-capable backends (object stores such as S3/Azure) are fed - // directly from the client stream through an io.Pipe, so CAS memory stays - // bounded by the chunk/pipe size regardless of artifact size (PFM-6923). - // The OCI backend, whose push path needs the whole layer content up front, - // does not advertise streaming and keeps the fully-buffered path. + // Streaming-capable backends (the object stores S3/Azure and the OCI backend) + // are fed directly from the client stream through an io.Pipe, so CAS memory + // stays bounded by the chunk/pipe size regardless of artifact size + // (PFM-6923). Backends that do not advertise streaming keep the + // fully-buffered path. var committedSize int64 if su, ok := storageBackend.(backend.StreamingUploader); ok && su.SupportsStreaming() { committedSize, err = s.streamUpload(ctx, stream, storageBackend, req, info.MaxBytes) @@ -166,11 +166,11 @@ func (s *ByteStreamService) Write(stream bytestream.ByteStream_WriteServer) erro } // bufferedUpload accumulates the whole artifact in memory before handing it to -// the backend. This is required by the OCI backend: its push implementation -// does not support streaming/chunked uploads for uncompressed layers (we can not -// use stream.Layer since it only supports compressed layers, and we want to -// store raw data with custom mimetypes), so it needs the full content up front. -// https://github.com/google/go-containerregistry/blob/main/pkg/v1/stream/README.md +// the backend. It is the fallback for backends that do not implement +// StreamingUploader. Streaming backends always take streamUpload (they report +// SupportsStreaming()==true unconditionally); when such a backend cannot stream +// a particular upload — e.g. the OCI backend when the size is unknown — it +// buffers internally, not through this path. // It returns the total number of bytes committed to the backend. Feed errors are // returned unwrapped (classified by the caller); backend Upload failures are // wrapped in backendUploadError so the caller always masks them. diff --git a/pkg/blobmanager/backend.go b/pkg/blobmanager/backend.go index c4c4f355a..fb525fccc 100644 --- a/pkg/blobmanager/backend.go +++ b/pkg/blobmanager/backend.go @@ -51,8 +51,8 @@ type UploaderDownloader interface { // The Artifact CAS service type-asserts uploaders against this interface: when // a backend reports SupportsStreaming()==true the upload is piped straight from // the client stream to the backend, bounding CAS memory usage independently of -// artifact size (PFM-6923). Backends that do not implement it (e.g. the OCI -// backend, whose push path needs the full layer content up front) keep the +// artifact size (PFM-6923). The object-store backends (S3, Azure) and +// the OCI backend implement it. A backend that does not implement it keeps the // buffered code path. type StreamingUploader interface { // SupportsStreaming reports whether Upload can be fed a streaming reader diff --git a/pkg/blobmanager/oci/backend.go b/pkg/blobmanager/oci/backend.go index 53b266e35..4a99e46b2 100644 --- a/pkg/blobmanager/oci/backend.go +++ b/pkg/blobmanager/oci/backend.go @@ -1,5 +1,5 @@ // -// Copyright 2023 The Chainloop Authors. +// Copyright 2023-2026 The Chainloop Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -16,6 +16,7 @@ package oci import ( + "bufio" "context" "errors" "fmt" @@ -102,23 +103,27 @@ func (b *Backend) Exists(_ context.Context, digest string) (bool, error) { return true, nil } -func (b *Backend) Upload(_ context.Context, r io.Reader, resource *pb.CASResource) error { - // We need to read the whole content before uploading it to the registry - // This is due to the fact that our OCI push implementation does not support streaming/chunks for uncompressed layers - // We can not use stream.Layer since it only supports compressed layers, we want to store raw data and set custom mimetypes - // https://github.com/google/go-containerregistry/blob/main/pkg/v1/stream/README.md - // TODO: Split content in multiple layers and do concurrent uploads/downloads - data, err := io.ReadAll(r) - if err != nil { - return fmt.Errorf("reading content: %w", err) - } +// sniffLen is the number of leading bytes http.DetectContentType inspects to +// determine the media type. We peek exactly this many bytes off the stream so +// the media type is detected identically to the buffered path without reading +// the whole artifact into memory. +const sniffLen = 512 +// SupportsStreaming reports that the OCI backend can upload directly from a +// streaming reader. Streaming requires the artifact size to be known up front +// (registry blobs are pushed with a fixed Content-Length); when the client did +// not provide it Upload transparently falls back to buffering the content. +func (b *Backend) SupportsStreaming() bool { return true } + +func (b *Backend) Upload(ctx context.Context, r io.Reader, resource *pb.CASResource) error { ref, err := name.ParseReference(b.resourcePath(resource.Digest)) if err != nil { return fmt.Errorf("parsing reference: %w", err) } - img, err := craftImage(data, resource) + // The image is either backed by a streaming layer (when the size is known) + // or a fully buffered one (legacy clients that do not send the size). + img, err := b.craftImageFromReader(r, resource) if err != nil { return fmt.Errorf("crafting image: %w", err) } @@ -127,23 +132,96 @@ func (b *Backend) Upload(_ context.Context, r io.Reader, resource *pb.CASResourc return fmt.Errorf("validating image: %w", err) } - err = remote.Write(ref, img, remote.WithAuthFromKeychain(b.keychain)) - if err != nil { + // Disable go-containerregistry's request retries: a streamed layer body + // cannot be replayed, so a retry would re-request the already-drained reader + // and upload truncated content. On a transient failure the whole upload is + // retried by the client instead. The retry-disable is harmless for the + // buffered path too. + if err := remote.Write(ref, img, + remote.WithAuthFromKeychain(b.keychain), + remote.WithContext(ctx), + remote.WithRetryPredicate(func(error) bool { return false }), + ); err != nil { return fmt.Errorf("writing image: %w", err) } return nil } +// craftImageFromReader builds the chainloop OCI image around the artifact +// content. When the artifact size is known (resource.Size > 0) the layer is +// streamed straight from r, keeping memory bounded regardless of artifact size. +// Otherwise it falls back to buffering the whole content, which is +// required for legacy clients that do not report the size (an empty artifact has +// size 0 and is rejected by the buffered path). +func (b *Backend) craftImageFromReader(r io.Reader, resource *pb.CASResource) (v1.Image, error) { + if resource == nil || resource.Digest == "" { + return nil, errors.New("resource metadata is not valid") + } + + if resource.Size <= 0 { + data, err := io.ReadAll(r) + if err != nil { + return nil, fmt.Errorf("reading content: %w", err) + } + return craftImage(data, resource) + } + + return craftStreamingImage(r, resource) +} + func (b *Backend) resourcePath(resourceName string) string { return fmt.Sprintf("%s/%s-%s", b.repo, b.prefix, resourceName) } +// craftStreamingImage assembles the image with a layer whose bytes are streamed +// from r. The media type is detected from the leading bytes, which are peeked +// (not consumed) so the streamed content is complete. +func craftStreamingImage(r io.Reader, resource *pb.CASResource) (v1.Image, error) { + hash, err := v1.NewHash("sha256:" + resource.Digest) + if err != nil { + return nil, fmt.Errorf("parsing digest: %w", err) + } + + // Peek the leading bytes to detect the media type without consuming them. + br := bufio.NewReaderSize(r, sniffLen) + head, err := br.Peek(sniffLen) + if err != nil && !errors.Is(err, io.EOF) && !errors.Is(err, bufio.ErrBufferFull) { + return nil, fmt.Errorf("detecting media type: %w", err) + } + + // Verify the streamed content against the declared digest as it flows to the + // registry. The buffered path got this for free (the layer digest was + // computed from the bytes); here rawStreamLayer.Digest() returns the declared + // digest, so without this a client could store content under a mismatched key + // on a registry that does not strictly verify blob digests on commit. + layer := &rawStreamLayer{ + hash: hash, + size: resource.Size, + mediaType: backend.DetectedMediaType(head), + body: io.NopCloser(newVerifyingReader(br, resource.Digest, resource.Size)), + } + + return buildImage(layer, resource) +} + +// craftImage builds the image from the whole content buffered in memory. Kept +// for legacy clients that do not report the artifact size up front. func craftImage(content []byte, resource *pb.CASResource) (v1.Image, error) { if len(content) == 0 { return nil, errors.New("content is empty") } + layer := static.NewLayer(content, backend.DetectedMediaType(content)) + return buildImage(layer, resource) +} + +// buildImage assembles the chainloop OCI image (OCI manifest + OCI config + a +// single raw layer + authors/title annotations) around an already-constructed +// layer. Using an OCIConfigJSON config makes mutate populate the config's +// rootfs.diff_ids from the layer DiffID, which the download path relies on +// (remote.Image().LayerByDiffID). +func buildImage(layer v1.Layer, resource *pb.CASResource) (v1.Image, error) { if resource == nil || resource.FileName == "" || resource.Digest == "" { return nil, errors.New("resource metadata is not valid") } @@ -156,7 +234,6 @@ func craftImage(content []byte, resource *pb.CASResource) (v1.Image, error) { ocispec.AnnotationTitle: resource.FileName, }).(v1.Image) - layer := static.NewLayer(content, backend.DetectedMediaType(content)) img, err := mutate.Append(base, mutate.Addendum{Layer: layer}) if err != nil { return nil, err diff --git a/pkg/blobmanager/oci/backend_streaming_test.go b/pkg/blobmanager/oci/backend_streaming_test.go new file mode 100644 index 000000000..3ac5defc6 --- /dev/null +++ b/pkg/blobmanager/oci/backend_streaming_test.go @@ -0,0 +1,127 @@ +// +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package oci + +import ( + "bytes" + "context" + "testing" + + pb "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1" + v1 "github.com/google/go-containerregistry/pkg/v1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// digestOf returns the sha256 hex digest of content, as stored by chainloop +// (no algorithm prefix). +func digestOf(content []byte) string { + h, _, err := v1.SHA256(bytes.NewReader(content)) + if err != nil { + panic(err) + } + return h.Hex +} + +// TestStreamingRoundTrip uploads content through the streaming path (size known) +// and downloads it back, asserting the bytes and metadata survive intact and the +// existing (untouched) download path keeps working. +func (s *testSuite) TestStreamingRoundTrip() { + testCases := []struct { + name string + content []byte + }{ + {name: "small text below sniff length", content: []byte("hello world")}, + {name: "empty-ish single byte", content: []byte("x")}, + // Larger than sniffLen (512) to exercise the peek + streamed remainder. + {name: "binary larger than sniff length", content: bytes.Repeat([]byte{0x00, 0x01, 0x02, 0x03}, 4096)}, + {name: "text larger than sniff length", content: bytes.Repeat([]byte("chainloop "), 1000)}, + } + + for _, tc := range testCases { + s.T().Run(tc.name, func(t *testing.T) { + digest := digestOf(tc.content) + resource := &pb.CASResource{ + Digest: digest, + FileName: "artifact.bin", + Size: int64(len(tc.content)), + } + + // Streamed upload (size > 0). + err := s.validBackend.Upload(context.Background(), bytes.NewReader(tc.content), resource) + require.NoError(t, err) + + // Download back through the existing path and compare. + var buf bytes.Buffer + err = s.validBackend.Download(context.Background(), &buf, digest) + require.NoError(t, err) + assert.Equal(t, tc.content, buf.Bytes()) + + // Describe must report the right size and filename. + desc, err := s.validBackend.Describe(context.Background(), digest) + require.NoError(t, err) + assert.Equal(t, digest, desc.Digest) + assert.Equal(t, "artifact.bin", desc.FileName) + assert.Equal(t, int64(len(tc.content)), desc.Size) + + exists, err := s.validBackend.Exists(context.Background(), digest) + require.NoError(t, err) + assert.True(t, exists) + }) + } +} + +// TestStreamingDigestMismatchRejected ensures the streaming path keeps the +// content-integrity guarantee the buffered path had: if the declared digest +// does not match the streamed bytes, the upload fails and nothing is committed. +func (s *testSuite) TestStreamingDigestMismatchRejected() { + content := bytes.Repeat([]byte("streamed content "), 100) + wrongDigest := digestOf([]byte("totally different content")) + resource := &pb.CASResource{ + Digest: wrongDigest, + FileName: "mismatch.bin", + Size: int64(len(content)), + } + + err := s.validBackend.Upload(context.Background(), bytes.NewReader(content), resource) + require.Error(s.T(), err) + assert.ErrorContains(s.T(), err, "content digest mismatch") + + // Nothing should have been committed under the (wrong) digest. + exists, err := s.validBackend.Exists(context.Background(), wrongDigest) + require.NoError(s.T(), err) + assert.False(s.T(), exists) +} + +// TestLegacyBufferedRoundTrip covers old clients that do not report the size: +// the upload falls back to buffering and still round-trips. +func (s *testSuite) TestLegacyBufferedRoundTrip() { + content := bytes.Repeat([]byte("legacy "), 500) + digest := digestOf(content) + resource := &pb.CASResource{ + Digest: digest, + FileName: "legacy.bin", + // Size intentionally left as 0 to emulate a legacy client. + } + + err := s.validBackend.Upload(context.Background(), bytes.NewReader(content), resource) + require.NoError(s.T(), err) + + var buf bytes.Buffer + err = s.validBackend.Download(context.Background(), &buf, digest) + require.NoError(s.T(), err) + assert.Equal(s.T(), content, buf.Bytes()) +} diff --git a/pkg/blobmanager/oci/backend_test.go b/pkg/blobmanager/oci/backend_test.go index c0d47638d..54f8535c1 100644 --- a/pkg/blobmanager/oci/backend_test.go +++ b/pkg/blobmanager/oci/backend_test.go @@ -48,7 +48,10 @@ func (s *testSuite) TestUpload() { }) s.T().Run("empty content", func(t *testing.T) { - err := s.validBackend.Upload(context.TODO(), bytes.NewBuffer(nil), s.casResource) + // An empty artifact has size 0, which routes through the buffered path + // that rejects empty content. + emptyResource := &pb.CASResource{Digest: s.casResource.Digest, FileName: s.casResource.FileName, Size: 0} + err := s.validBackend.Upload(context.TODO(), bytes.NewBuffer(nil), emptyResource) assert.Error(s.T(), err) assert.ErrorContains(s.T(), err, "content is empty") }) @@ -410,13 +413,13 @@ func TestOCIBackend(t *testing.T) { suite.Run(t, new(testSuite)) } -// TestBackend_DoesNotSupportStreaming pins the OCI backend as NON-streaming. -// go-containerregistry's push path needs the whole layer content in memory up -// front (see Backend.Upload), so the CAS service must keep buffering OCI -// uploads. If OCI ever grows a SupportsStreaming method it would be fed a -// streaming reader and silently break; this test fails closed against that. -func TestBackend_DoesNotSupportStreaming(t *testing.T) { +// TestBackend_SupportsStreaming pins the OCI backend as streaming-capable: it +// implements backend.StreamingUploader and reports true, so the CAS service +// feeds it directly from the client stream and CAS memory stays bounded +// regardless of artifact size. +func TestBackend_SupportsStreaming(t *testing.T) { var b backend.UploaderDownloader = &Backend{} - _, ok := b.(backend.StreamingUploader) - require.False(t, ok, "oci backend must NOT implement backend.StreamingUploader; it requires full in-memory buffering") + su, ok := b.(backend.StreamingUploader) + require.True(t, ok, "oci backend must implement backend.StreamingUploader") + require.True(t, su.SupportsStreaming()) } diff --git a/pkg/blobmanager/oci/stream_layer.go b/pkg/blobmanager/oci/stream_layer.go new file mode 100644 index 000000000..56b89367c --- /dev/null +++ b/pkg/blobmanager/oci/stream_layer.go @@ -0,0 +1,129 @@ +// +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package oci + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "hash" + "io" + "sync/atomic" + + v1 "github.com/google/go-containerregistry/pkg/v1" + "github.com/google/go-containerregistry/pkg/v1/types" +) + +// rawStreamLayer is a v1.Layer implementation for content-addressed raw +// (uncompressed) blobs whose digest and size are known up front. It lets the +// artifact bytes be streamed straight to the registry blob endpoint without +// buffering the whole content in memory. +// +// go-containerregistry's remote.Write passes the value returned by Compressed() +// directly as the HTTP request body (no io.ReadAll), so a Compressed() that +// yields the raw stream streams the artifact through with bounded memory. The +// provided stream.Layer cannot be used here because it gzip-compresses the +// content and only computes the digest after the stream is consumed, which is +// incompatible with storing raw, content-addressed data. +// +// Because the content is stored uncompressed, the layer Digest and DiffID are +// identical; mutate embeds the DiffID into the image config's rootfs.diff_ids, +// keeping the existing download path (remote.Image().LayerByDiffID) intact. +// +// The underlying stream can be read only once. Compressed/Uncompressed return an +// error on any subsequent call so that a retry or an HTTP/2 body replay fails +// cleanly instead of silently uploading a truncated, already-drained reader. +type rawStreamLayer struct { + hash v1.Hash + size int64 + mediaType types.MediaType + body io.ReadCloser + consumed atomic.Bool +} + +// Digest returns the sha256 of the raw content, known up front. +func (l *rawStreamLayer) Digest() (v1.Hash, error) { return l.hash, nil } + +// DiffID equals Digest because the content is stored uncompressed. +func (l *rawStreamLayer) DiffID() (v1.Hash, error) { return l.hash, nil } + +// Size returns the raw content size in bytes, known up front. +func (l *rawStreamLayer) Size() (int64, error) { return l.size, nil } + +// MediaType returns the detected media type of the raw content. +func (l *rawStreamLayer) MediaType() (types.MediaType, error) { return l.mediaType, nil } + +// Compressed returns the raw content stream. The content is not compressed; the +// name follows the v1.Layer interface, and remote.Write sends these bytes to the +// registry verbatim. +func (l *rawStreamLayer) Compressed() (io.ReadCloser, error) { return l.reader() } + +// Uncompressed returns the same raw content stream as Compressed. +func (l *rawStreamLayer) Uncompressed() (io.ReadCloser, error) { return l.reader() } + +func (l *rawStreamLayer) reader() (io.ReadCloser, error) { + if l.consumed.Swap(true) { + return nil, errors.New("raw stream layer body already consumed: a streamed upload cannot be retried on a non-replayable reader") + } + return l.body, nil +} + +// verifyingReader streams r while computing its sha256 over the first size bytes +// and fails if the digest does not match wantHex. It restores the +// content-integrity check the buffered OCI upload path performed (static.NewLayer +// computed the layer digest from the bytes, which validateImage then compared to +// the declared digest) without buffering the artifact: a mismatch makes Read +// return an error, which aborts remote.Write before the blob is committed. +// +// The check fires as soon as size bytes have been read, not only at EOF: the +// HTTP transport uploads exactly Content-Length (size) bytes and may stop +// reading there without ever pulling the trailing EOF, so an EOF-only check +// could be skipped entirely before the registry commit. +type verifyingReader struct { + r io.Reader + hasher hash.Hash + wantHex string + remaining int64 + verified bool +} + +func newVerifyingReader(r io.Reader, wantHex string, size int64) *verifyingReader { + return &verifyingReader{r: r, hasher: sha256.New(), wantHex: wantHex, remaining: size} +} + +func (v *verifyingReader) Read(p []byte) (int, error) { + n, err := v.r.Read(p) + if n > 0 && v.remaining > 0 { + // Hash at most the declared number of bytes: only those are uploaded. + h := int64(n) + if h > v.remaining { + h = v.remaining + } + // hash.Hash never returns an error. + _, _ = v.hasher.Write(p[:h]) + v.remaining -= h + } + // Verify once the declared byte count is reached or at EOF, whichever comes + // first. + if !v.verified && (v.remaining <= 0 || errors.Is(err, io.EOF)) { + v.verified = true + if got := hex.EncodeToString(v.hasher.Sum(nil)); got != v.wantHex { + return n, fmt.Errorf("content digest mismatch: got sha256:%s, want sha256:%s", got, v.wantHex) + } + } + return n, err +} diff --git a/pkg/blobmanager/oci/stream_layer_test.go b/pkg/blobmanager/oci/stream_layer_test.go new file mode 100644 index 000000000..d28d2ca24 --- /dev/null +++ b/pkg/blobmanager/oci/stream_layer_test.go @@ -0,0 +1,121 @@ +// +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package oci + +import ( + "bytes" + "io" + "strings" + "testing" + + v1 "github.com/google/go-containerregistry/pkg/v1" + "github.com/google/go-containerregistry/pkg/v1/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRawStreamLayer_Metadata(t *testing.T) { + content := "hello world" + hash, _, err := v1.SHA256(strings.NewReader(content)) + require.NoError(t, err) + + l := &rawStreamLayer{ + hash: hash, + size: int64(len(content)), + mediaType: types.MediaType("text/plain"), + body: io.NopCloser(strings.NewReader(content)), + } + + gotDigest, err := l.Digest() + require.NoError(t, err) + assert.Equal(t, hash, gotDigest) + + // Content is stored uncompressed, so DiffID must equal Digest; this is what + // keeps the config rootfs.diff_ids consistent for the download path. + gotDiffID, err := l.DiffID() + require.NoError(t, err) + assert.Equal(t, hash, gotDiffID) + + gotSize, err := l.Size() + require.NoError(t, err) + assert.Equal(t, int64(len(content)), gotSize) + + gotMT, err := l.MediaType() + require.NoError(t, err) + assert.Equal(t, types.MediaType("text/plain"), gotMT) +} + +func TestRawStreamLayer_BodyStreamedRaw(t *testing.T) { + content := "the quick brown fox" + l := &rawStreamLayer{body: io.NopCloser(strings.NewReader(content))} + + // Compressed returns the raw bytes verbatim (no gzip). + rc, err := l.Compressed() + require.NoError(t, err) + got, err := io.ReadAll(rc) + require.NoError(t, err) + assert.Equal(t, content, string(got)) +} + +func TestVerifyingReader(t *testing.T) { + content := []byte("the quick brown fox") + digest := func(b []byte) string { + h, _, err := v1.SHA256(bytes.NewReader(b)) + require.NoError(t, err) + return h.Hex + } + + t.Run("matching digest streams through cleanly", func(t *testing.T) { + vr := newVerifyingReader(bytes.NewReader(content), digest(content), int64(len(content))) + got, err := io.ReadAll(vr) + require.NoError(t, err) + assert.Equal(t, content, got) + }) + + t.Run("mismatching digest errors", func(t *testing.T) { + // Declare the digest of different content than what actually streams. + vr := newVerifyingReader(bytes.NewReader(content), digest([]byte("something else")), int64(len(content))) + _, err := io.ReadAll(vr) + require.Error(t, err) + assert.ErrorContains(t, err, "content digest mismatch") + }) + + t.Run("verifies at the declared byte count without a trailing EOF", func(t *testing.T) { + // Mimic an HTTP transport bounded by Content-Length: copy exactly + // len(content) bytes through a LimitReader, which stops there and never + // pulls the underlying EOF. The mismatch must still be caught. + vr := newVerifyingReader(bytes.NewReader(content), digest([]byte("other content")), int64(len(content))) + _, err := io.Copy(io.Discard, io.LimitReader(vr, int64(len(content)))) + require.Error(t, err) + assert.ErrorContains(t, err, "content digest mismatch") + }) +} + +func TestRawStreamLayer_ConsumableOnce(t *testing.T) { + l := &rawStreamLayer{body: io.NopCloser(strings.NewReader("data"))} + + _, err := l.Compressed() + require.NoError(t, err) + + // A second read (retry / HTTP body replay) must fail cleanly rather than + // return a drained reader that would upload truncated content. + _, err = l.Compressed() + require.Error(t, err) + assert.ErrorContains(t, err, "already consumed") + + _, err = l.Uncompressed() + require.Error(t, err) +} diff --git a/pkg/casclient/uploader.go b/pkg/casclient/uploader.go index 550e66ccd..e75bae6b4 100644 --- a/pkg/casclient/uploader.go +++ b/pkg/casclient/uploader.go @@ -1,5 +1,5 @@ // -// Copyright 2023-2025 The Chainloop Authors. +// Copyright 2023-2026 The Chainloop Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -25,6 +25,7 @@ import ( "io" "os" "path" + "strings" "code.cloudfoundry.org/bytefmt" v1 "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1" @@ -69,7 +70,13 @@ func (c *Client) Upload(ctx context.Context, r io.Reader, filename, digest strin ctx, cancel := context.WithCancel(ctx) defer cancel() - resource, err := encodeResource(filename, h.String()) + // Detecting the content size lets the CAS stream the upload straight to + // object-store/OCI backends with a bounded memory footprint. When the size + // cannot be derived without consuming the reader it is left as 0 and the CAS + // transparently falls back to buffering. + size, _ := readerLen(r) + + resource, err := encodeResource(filename, h.String(), size) if err != nil { return nil, fmt.Errorf("encoding resource name: %w", err) } @@ -159,8 +166,11 @@ doUpload: return latestStatus, nil } -// encodedResource returns a base64-encoded v1.UploadResource which wraps both the digest and fileName -func encodeResource(fileName, digest string) (string, error) { +// encodedResource returns a base64-encoded v1.UploadResource which wraps the +// digest, fileName and (when known) the content size. The size lets the CAS +// stream the upload to its backend; a zero size means "unknown" and the CAS +// buffers instead. +func encodeResource(fileName, digest string, size int64) (string, error) { if fileName == "" { return "", fmt.Errorf("file name is empty") } @@ -174,7 +184,7 @@ func encodeResource(fileName, digest string) (string, error) { var encodedResource bytes.Buffer enc := gob.NewEncoder(&encodedResource) // Currently we only support SHA256 - r := &v1.CASResource{FileName: fileName, Digest: h.Hex} + r := &v1.CASResource{FileName: fileName, Digest: h.Hex, Size: size} if err := enc.Encode(r); err != nil { return "", err @@ -182,3 +192,32 @@ func encodeResource(fileName, digest string) (string, error) { return base64.StdEncoding.EncodeToString(encodedResource.Bytes()), nil } + +// readerLen derives the number of bytes r will yield without consuming it, +// mirroring how net/http derives a request body's ContentLength. It returns +// false when the size cannot be determined, in which case the caller must treat +// the size as unknown. The reader position is not modified. +func readerLen(r io.Reader) (int64, bool) { + switch v := r.(type) { + case *bytes.Buffer: + return int64(v.Len()), true + case *bytes.Reader: + return int64(v.Len()), true + case *strings.Reader: + return int64(v.Len()), true + case *os.File: + info, err := v.Stat() + if err != nil || !info.Mode().IsRegular() { + return 0, false + } + // Account for any bytes already consumed so the reported size matches + // what remains to be read (and hashed). + cur, err := v.Seek(0, io.SeekCurrent) + if err != nil { + return 0, false + } + return info.Size() - cur, true + default: + return 0, false + } +} diff --git a/pkg/casclient/uploader_test.go b/pkg/casclient/uploader_test.go index d277b320b..cd8423cba 100644 --- a/pkg/casclient/uploader_test.go +++ b/pkg/casclient/uploader_test.go @@ -1,5 +1,5 @@ // -// Copyright 2023-2025 The Chainloop Authors. +// Copyright 2023-2026 The Chainloop Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -20,10 +20,14 @@ import ( "encoding/base64" "encoding/gob" "fmt" + "io" + "os" + "strings" "testing" v1 "github.com/chainloop-dev/chainloop/app/artifact-cas/api/cas/v1" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestEncodeResource(t *testing.T) { @@ -33,6 +37,7 @@ func TestEncodeResource(t *testing.T) { name string fileName string digest string + size int64 want *v1.CASResource wantError bool }{ @@ -64,11 +69,18 @@ func TestEncodeResource(t *testing.T) { fileName: "foo.txt", want: &v1.CASResource{FileName: "foo.txt", Digest: validDigestHex}, }, + { + name: "valid with size", + digest: fmt.Sprintf("sha256:%s", validDigestHex), + fileName: "foo.txt", + size: 1234, + want: &v1.CASResource{FileName: "foo.txt", Digest: validDigestHex, Size: 1234}, + }, } for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - gotEncoded, err := encodeResource(tc.fileName, tc.digest) + gotEncoded, err := encodeResource(tc.fileName, tc.digest, tc.size) if tc.wantError { assert.Error(t, err) return @@ -87,3 +99,51 @@ func TestEncodeResource(t *testing.T) { }) } } + +func TestReaderLen(t *testing.T) { + const content = "the quick brown fox" + + t.Run("bytes.Reader", func(t *testing.T) { + n, ok := readerLen(bytes.NewReader([]byte(content))) + assert.True(t, ok) + assert.Equal(t, int64(len(content)), n) + }) + + t.Run("bytes.Buffer", func(t *testing.T) { + n, ok := readerLen(bytes.NewBufferString(content)) + assert.True(t, ok) + assert.Equal(t, int64(len(content)), n) + }) + + t.Run("strings.Reader", func(t *testing.T) { + n, ok := readerLen(strings.NewReader(content)) + assert.True(t, ok) + assert.Equal(t, int64(len(content)), n) + }) + + t.Run("os.File reports remaining bytes", func(t *testing.T) { + f, err := os.CreateTemp(t.TempDir(), "readerlen") + require.NoError(t, err) + _, err = f.WriteString(content) + require.NoError(t, err) + _, err = f.Seek(0, io.SeekStart) + require.NoError(t, err) + + n, ok := readerLen(f) + assert.True(t, ok) + assert.Equal(t, int64(len(content)), n) + + // After consuming some bytes, the remaining size is reported. + buf := make([]byte, 4) + _, err = io.ReadFull(f, buf) + require.NoError(t, err) + n, ok = readerLen(f) + assert.True(t, ok) + assert.Equal(t, int64(len(content)-4), n) + }) + + t.Run("unknown reader is undetectable", func(t *testing.T) { + _, ok := readerLen(io.NopCloser(strings.NewReader(content))) + assert.False(t, ok) + }) +}