Bound external reference reads and fingerprints

internal/framework/pipeline/handoff.go

@@ -243,6 +243,7 @@ type generatedReferenceFingerprintIdentity struct {
 	SchemaDigest     string `json:"schema_digest"`
 	MediaType        string `json:"media_type"`
 	ContentDigest    string `json:"content_digest"`
+	SizeBytes        int64  `json:"size_bytes"`
 }
 
 // generatedReferenceDependencies returns the canonical semantic dependency
@@ -271,6 +272,7 @@ func generatedReferenceDependencies(set contracts.ReferenceSet) []CheckpointFing
 				SchemaDigest:     contracts.DigestArtifactSchema(item.ArtifactSchema),
 				MediaType:        item.MediaType,
 				ContentDigest:    item.Digest,
+				SizeBytes:        item.SizeBytes,
 			}
 			encoded, err := json.Marshal(identity)
 			if err != nil {

internal/framework/pipeline/handoff_test.go

@@ -185,6 +185,27 @@ func TestGeneratedReferenceFingerprintChangesWithCanonicalContent(t *testing.T)
 	}
 }
 
+func TestGeneratedReferenceFingerprintChangesWithCanonicalSize(t *testing.T) {
+	input, step, producerOutput := handoffFixture(t, codecNotes{Items: []string{"first"}})
+	sets, _, err := buildStepReferenceSets(input, step, []contracts.SerializedOutput{producerOutput})
+	if err != nil {
+		t.Fatalf("build reference set: %v", err)
+	}
+	key := keyForReferenceTarget(step.lanes[0].resolved.ExtractReferences)
+	firstDeps := generatedReferenceDependencies(sets[key])
+	changed := CloneReferenceSet(sets[key])
+	item := changed.Slots["producer-output"].Items[0]
+	item.SizeBytes++
+	changed.Slots["producer-output"] = contracts.ResolvedReferenceSlot{
+		Slot:  changed.Slots["producer-output"].Slot,
+		Items: []contracts.ReferenceItem{item},
+	}
+	secondDeps := generatedReferenceDependencies(changed)
+	if reflect.DeepEqual(firstDeps, secondDeps) {
+		t.Fatalf("generated dependencies = %#v, want size-sensitive fingerprint", firstDeps)
+	}
+}
+
 func TestRunnerHandsOffAcceptedNormalizedOutputBeforeConsumerLanes(t *testing.T) {
 	input, _, _ := handoffFixture(t, codecNotes{Items: []string{"first"}})
 	prepared := input.Prepared

internal/framework/pipeline/references.go

@@ -3,7 +3,10 @@ package pipeline
 import (
 	"crypto/sha256"
 	"encoding/hex"
+	"errors"
 	"fmt"
+	"io"
+	"math"
 	"mime"
 	"net/url"
 	"os"
@@ -124,10 +127,30 @@ func materializeReferenceTarget(
 		if err != nil {
 			return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q: %w", referenceTargetContext(pipelineID, target), slotName, binding.Source, err)
 		}
-		content, err := os.ReadFile(path)
+		file, err := os.Open(path)
 		if err != nil {
 			return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q read %q: %w", referenceTargetContext(pipelineID, target), slotName, path, err)
 		}
+		regularFileSize := int64(0)
+		if info, statErr := file.Stat(); statErr == nil && info.Mode().IsRegular() {
+			regularFileSize = info.Size()
+		}
+		content, err := readReferenceContent(file, slot.MaxBytes)
+		closeErr := file.Close()
+		if err == nil && closeErr != nil {
+			err = closeErr
+		}
+		if err != nil {
+			var sizeErr *referenceSizeLimitError
+			if errors.As(err, &sizeErr) {
+				sizeBytes := sizeErr.SizeBytes
+				if regularFileSize > sizeBytes {
+					sizeBytes = regularFileSize
+				}
+				return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q is %d bytes, limit %d", referenceTargetContext(pipelineID, target), slotName, path, sizeBytes, slot.MaxBytes)
+			}
+			return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q read %q: %w", referenceTargetContext(pipelineID, target), slotName, path, err)
+		}
 		if !utf8.Valid(content) {
 			return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q must be UTF-8 text", referenceTargetContext(pipelineID, target), slotName, path)
 		}
@@ -135,9 +158,6 @@ func materializeReferenceTarget(
 		if !referenceMediaTypeAccepted(mediaType, slot.AcceptedMediaTypes) {
 			return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q media type %q is not accepted", referenceTargetContext(pipelineID, target), slotName, path, mediaType)
 		}
-		if slot.MaxBytes > 0 && int64(len(content)) > slot.MaxBytes {
-			return contracts.ReferenceSet{}, nil, fmt.Errorf("%s reference slot %q path %q is %d bytes, limit %d", referenceTargetContext(pipelineID, target), slotName, path, len(content), slot.MaxBytes)
-		}
 		if len(content) == 0 {
 			warnings = append(warnings, contracts.Warning{
 				Scope:      referenceWarningScope(pipelineID, target, slotName),
@@ -163,6 +183,32 @@ func materializeReferenceTarget(
 	return set, warnings, nil
 }
 
+type referenceSizeLimitError struct {
+	SizeBytes int64
+	MaxBytes  int64
+}
+
+func (err *referenceSizeLimitError) Error() string {
+	return fmt.Sprintf("reference is %d bytes, limit %d", err.SizeBytes, err.MaxBytes)
+}
+
+func readReferenceContent(reader io.Reader, maxBytes int64) ([]byte, error) {
+	if maxBytes <= 0 {
+		return io.ReadAll(reader)
+	}
+	if maxBytes == math.MaxInt64 {
+		return nil, fmt.Errorf("maximum reference size %d cannot be safely bounded", maxBytes)
+	}
+	content, err := io.ReadAll(io.LimitReader(reader, maxBytes+1))
+	if err != nil {
+		return nil, err
+	}
+	if int64(len(content)) > maxBytes {
+		return nil, &referenceSizeLimitError{SizeBytes: int64(len(content)), MaxBytes: maxBytes}
+	}
+	return content, nil
+}
+
 func referenceTargetSpec(target ResolvedReferenceTarget, artifactKind contracts.ArtifactKind, catalog ModuleCatalog) (ModuleSpec, error) {
 	switch target.Stage {
 	case StageChunk:

internal/framework/pipeline/references_test.go

@@ -2,6 +2,9 @@ package pipeline
 
 import (
 	"encoding/json"
+	"errors"
+	"io"
+	"math"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -12,6 +15,61 @@ import (
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
 )
 
+type countingReader struct {
+	content  []byte
+	consumed int
+}
+
+func (reader *countingReader) Read(buffer []byte) (int, error) {
+	if len(reader.content) == 0 {
+		return 0, io.EOF
+	}
+	read := copy(buffer, reader.content)
+	reader.content = reader.content[read:]
+	reader.consumed += read
+	return read, nil
+}
+
+func TestReadReferenceContentBoundsPositiveMaximum(t *testing.T) {
+	tests := []struct {
+		name          string
+		content       string
+		maxBytes      int64
+		wantContent   string
+		wantConsumed  int
+		wantLimitErr  bool
+		wantReadError bool
+	}{
+		{name: "exact maximum", content: "abc", maxBytes: 3, wantContent: "abc", wantConsumed: 3},
+		{name: "oversized", content: "abcdef", maxBytes: 3, wantConsumed: 4, wantLimitErr: true},
+		{name: "unbounded", content: "abcdef", maxBytes: 0, wantContent: "abcdef", wantConsumed: 6},
+		{name: "negative maximum remains unbounded", content: "abcdef", maxBytes: -1, wantContent: "abcdef", wantConsumed: 6},
+		{name: "overflowing maximum", content: "abcdef", maxBytes: math.MaxInt64, wantConsumed: 0, wantReadError: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			reader := &countingReader{content: []byte(tt.content)}
+			content, err := readReferenceContent(reader, tt.maxBytes)
+			if reader.consumed != tt.wantConsumed {
+				t.Fatalf("reader consumed %d bytes, want %d", reader.consumed, tt.wantConsumed)
+			}
+			if got := string(content); got != tt.wantContent {
+				t.Fatalf("content = %q, want %q", got, tt.wantContent)
+			}
+			var sizeErr *referenceSizeLimitError
+			if errors.As(err, &sizeErr) != tt.wantLimitErr {
+				t.Fatalf("error = %v, want limit error=%t", err, tt.wantLimitErr)
+			}
+			if (err != nil) != (tt.wantLimitErr || tt.wantReadError) {
+				t.Fatalf("error = %v, want error=%t", err, tt.wantLimitErr || tt.wantReadError)
+			}
+			if sizeErr != nil && (sizeErr.SizeBytes != 4 || sizeErr.MaxBytes != 3) {
+				t.Fatalf("size error = %#v, want observed size 4 and limit 3", sizeErr)
+			}
+		})
+	}
+}
+
 func TestMaterializeReferencesResolvesPathsAndDigestsContent(t *testing.T) {
 	configDir := t.TempDir()
 	workingDir := t.TempDir()