amadan

carlos / platform Public

release: stream an object to a file with a running sha

Opener is a side-interface (48 test fakes implement Store; none break),
implemented by DirStore, S3Store, prefixStore and the fleet door client.
FetchToFile is what a box uses for a platform binary: limit+1 read, sha
checked before rename, directory fsync after.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015QCUCpfxbCyEAVQ2m51dH5
Paul Campbell pushed by paul@keymail.dev 3d4ef80880a26d8d60ce9e7df1c1a203387e1a16 parent c632c38
5 files changed, +287 −0
  • internal/fleetapi/fleetapi.go +26 −0
  • internal/fleetapi/fleetapi_test.go +27 −0
  • internal/release/fetch.go +70 −0
  • internal/release/fetch_test.go +105 −0
  • internal/release/store.go +59 −0
diff --git a/internal/fleetapi/fleetapi.go b/internal/fleetapi/fleetapi.go
index 73f1e0402..c21ea153d 100644
--- a/internal/fleetapi/fleetapi.go
+++ b/internal/fleetapi/fleetapi.go
@@ -154,6 +154,32 @@ func (s *store) Get(ctx context.Context, key string) ([]byte, string, error) {
}
}
+// Open streams one object. Unlike Get it does not buffer the body, which
+// is what a 25 MiB platform binary on a Pi needs; the 404 mapping is the
+// same as Get's so callers can errors.Is(err, os.ErrNotExist).
+func (s *store) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
+ req, err := s.newRequest(ctx, http.MethodGet, s.objectURL(key), nil)
+ if err != nil {
+ return nil, "", err
+ }
+ resp, err := s.client.Do(req)
+ if err != nil {
+ return nil, "", errorf("open %s: %w", key, err)
+ }
+ switch resp.StatusCode {
+ case http.StatusOK:
+ return resp.Body, resp.Header.Get("ETag"), nil
+ case http.StatusNotFound:
+ resp.Body.Close()
+ return nil, "", fmt.Errorf("fleetapi: %s: %w", key, os.ErrNotExist)
+ default:
+ resp.Body.Close()
+ return nil, "", errorf("open %s: unexpected status %d", key, resp.StatusCode)
+ }
+}
+
+var _ release.Opener = (*store)(nil)
+
// Put mirrors the door's CAS mapping exactly (storedoor.go's
// putPrecondition / F4): ifMatch == "" sends If-None-Match: * (create-only,
// the contract's only form of an unconditional-looking write), a non-empty
diff --git a/internal/fleetapi/fleetapi_test.go b/internal/fleetapi/fleetapi_test.go
index 4956c9b18..b372a666f 100644
--- a/internal/fleetapi/fleetapi_test.go
+++ b/internal/fleetapi/fleetapi_test.go
@@ -11,10 +11,12 @@ package fleetapi_test
// box dialing CARLOS_TUNNEL_URL would produce.
import (
+ "bytes"
"context"
"crypto/sha256"
"encoding/hex"
"errors"
+ "io"
"net"
"net/http"
"os"
@@ -364,3 +366,28 @@ func TestTokenNeverInErrorStrings(t *testing.T) {
}
}
}
+
+// ---- Open (streaming reads) ------------------------------------------------
+
+func TestOpenStreamsThroughTheDoor(t *testing.T) {
+ d := newTestDoor(t)
+ body := bytes.Repeat([]byte("x"), 1<<20)
+ seed(t, d.store, "bootstrap/carlos", string(body))
+ c := d.box(t)
+ op, ok := c.(release.Opener)
+ if !ok {
+ t.Fatal("fleetapi store is not an Opener")
+ }
+ rc, _, err := op.Open(context.Background(), "bootstrap/carlos")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer rc.Close()
+ got, _ := io.ReadAll(rc)
+ if !bytes.Equal(got, body) {
+ t.Fatalf("read %d bytes, want %d", len(got), len(body))
+ }
+ if _, _, err := op.Open(context.Background(), "bootstrap/nothing"); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("missing: %v", err)
+ }
+}
diff --git a/internal/release/fetch.go b/internal/release/fetch.go
new file mode 100644
index 000000000..4c3d33825
--- /dev/null
+++ b/internal/release/fetch.go
@@ -0,0 +1,70 @@
+package release
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+)
+
+// ErrTooLarge: the object exceeded the caller's limit. The reader is cut at
+// limit+1 bytes, so an object EXACTLY at the limit is told from one over it.
+var ErrTooLarge = errors.New("store: object larger than the caller's limit")
+
+// FetchToFile streams key into dst with a running sha256, never holding the
+// object in memory: temp file, fsync, rename, directory fsync. Any failure
+// removes the temp and leaves dst untouched.
+func FetchToFile(ctx context.Context, s Store, key, dst string, limit int64, wantSHA256 string) (err error) {
+ rc, _, err := OpenObject(ctx, s, key)
+ if err != nil {
+ return err
+ }
+ defer rc.Close()
+ tmp := dst + ".tmp"
+ f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600)
+ if err != nil {
+ return err
+ }
+ defer func() {
+ if err != nil {
+ f.Close()
+ os.Remove(tmp)
+ }
+ }()
+ h := sha256.New()
+ n, err := io.Copy(io.MultiWriter(f, h), io.LimitReader(rc, limit+1))
+ if err != nil {
+ return fmt.Errorf("fetch %s: %w", key, err)
+ }
+ if n > limit {
+ return fmt.Errorf("fetch %s: %w (limit %d)", key, ErrTooLarge, limit)
+ }
+ if got := hex.EncodeToString(h.Sum(nil)); got != wantSHA256 {
+ return fmt.Errorf("fetch %s: sha256 %s does not match the signed %s", key, got, wantSHA256)
+ }
+ if err = f.Sync(); err != nil {
+ return err
+ }
+ if err = f.Close(); err != nil {
+ return err
+ }
+ if err = os.Rename(tmp, dst); err != nil {
+ return err
+ }
+ return syncDir(filepath.Dir(dst))
+}
+
+// syncDir makes a rename durable: the directory entry is what fsync(file)
+// does not cover.
+func syncDir(dir string) error {
+ d, err := os.Open(dir)
+ if err != nil {
+ return err
+ }
+ defer d.Close()
+ return d.Sync()
+}
diff --git a/internal/release/fetch_test.go b/internal/release/fetch_test.go
new file mode 100644
index 000000000..a4cf3da53
--- /dev/null
+++ b/internal/release/fetch_test.go
@@ -0,0 +1,105 @@
+package release
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+)
+
+func shaOf(b []byte) string { s := sha256.Sum256(b); return hex.EncodeToString(s[:]) }
+
+func TestFetchToFileStreamsAndVerifies(t *testing.T) {
+ ctx := context.Background()
+ body := []byte(strings.Repeat("carlos", 10000))
+ for name, s := range map[string]Store{
+ "dir": NewDirStore(t.TempDir()),
+ "prefix": NewPrefixStore(NewDirStore(t.TempDir()), "fleets/pi/"),
+ } {
+ t.Run(name, func(t *testing.T) {
+ if _, err := s.Put(ctx, "bootstrap/router/v1/carlos-linux-arm64", body, ""); err != nil {
+ t.Fatal(err)
+ }
+ dst := filepath.Join(t.TempDir(), "carlos")
+ if err := FetchToFile(ctx, s, "bootstrap/router/v1/carlos-linux-arm64", dst, int64(len(body)), shaOf(body)); err != nil {
+ t.Fatal(err)
+ }
+ got, _ := os.ReadFile(dst)
+ if string(got) != string(body) {
+ t.Fatal("content differs")
+ }
+ if _, err := os.Stat(dst + ".tmp"); !errors.Is(err, os.ErrNotExist) {
+ t.Fatal("temp file left behind")
+ }
+ // One byte over the limit is refused, and nothing is left on disk.
+ dst2 := filepath.Join(t.TempDir(), "carlos")
+ if err := FetchToFile(ctx, s, "bootstrap/router/v1/carlos-linux-arm64", dst2, int64(len(body))-1, shaOf(body)); !errors.Is(err, ErrTooLarge) {
+ t.Fatalf("over-limit err = %v", err)
+ }
+ if _, err := os.Stat(dst2); !errors.Is(err, os.ErrNotExist) {
+ t.Fatal("over-limit fetch left a file")
+ }
+ // A sha mismatch is refused and leaves nothing.
+ dst3 := filepath.Join(t.TempDir(), "carlos")
+ err := FetchToFile(ctx, s, "bootstrap/router/v1/carlos-linux-arm64", dst3, int64(len(body)), strings.Repeat("0", 64))
+ if err == nil || !strings.Contains(err.Error(), "sha256") {
+ t.Fatalf("mismatch err = %v", err)
+ }
+ if _, err := os.Stat(dst3); !errors.Is(err, os.ErrNotExist) {
+ t.Fatal("mismatch fetch left a file")
+ }
+ // Missing key is ErrNotExist.
+ if err := FetchToFile(ctx, s, "bootstrap/router/v9/carlos-linux-arm64", dst3, 10, shaOf(nil)); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("missing err = %v", err)
+ }
+ })
+ }
+}
+
+func TestPrefixStoreOpenRebasesTheKey(t *testing.T) {
+ // prefixStore embeds the Store INTERFACE, so a promoted Open would forward
+ // the raw key and silently escape the prefix. It must be written out.
+ ctx := context.Background()
+ inner := NewDirStore(t.TempDir())
+ if _, err := inner.Put(ctx, "fleets/pi/x", []byte("in"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := inner.Put(ctx, "x", []byte("OUT"), ""); err != nil {
+ t.Fatal(err)
+ }
+ rc, _, err := OpenObject(ctx, NewPrefixStore(inner, "fleets/pi/"), "x")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer rc.Close()
+ b := make([]byte, 8)
+ n, _ := rc.Read(b)
+ if string(b[:n]) != "in" {
+ t.Fatalf("read %q through the prefix, want \"in\"", b[:n])
+ }
+}
+
+func TestOpenObjectFallsBackToGet(t *testing.T) {
+ // A Store that is not an Opener still works: OpenObject buffers via Get.
+ s := getOnly{NewDirStore(t.TempDir())}
+ if _, err := s.Put(context.Background(), "k", []byte("v"), ""); err != nil {
+ t.Fatal(err)
+ }
+ rc, _, err := OpenObject(context.Background(), s, "k")
+ if err != nil {
+ t.Fatal(err)
+ }
+ b := make([]byte, 4)
+ n, _ := rc.Read(b)
+ if string(b[:n]) != "v" {
+ t.Fatalf("got %q", b[:n])
+ }
+}
+
+type getOnly struct{ *DirStore }
+
+func (getOnly) Open() {} // different signature: getOnly is NOT an Opener
diff --git a/internal/release/store.go b/internal/release/store.go
index 3cdb23a1c..6f8c03e6f 100644
--- a/internal/release/store.go
+++ b/internal/release/store.go
@@ -100,6 +100,12 @@ func (p prefixStore) Get(ctx context.Context, key string) ([]byte, string, error
return p.Store.Get(ctx, p.full(key))
}
+// Open is written out, not promoted: the embedded field is the interface,
+// and a promoted Open would forward the UNREBASED key past the prefix.
+func (p prefixStore) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
+ return OpenObject(ctx, p.Store, p.full(key))
+}
+
func (p prefixStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
return p.Store.Put(ctx, p.full(key), body, ifMatch)
}
@@ -188,6 +194,33 @@ var (
_ SizeLister = (*S3Store)(nil)
)
+// Opener is the optional streaming read. Store.Get returns the whole object
+// in memory, which is fine for a pointer and wrong for a platform binary on
+// a box under MemoryHigh=70%. Implemented by every production store; a test
+// fake that only has Get still works through OpenObject. The etag may be ""
+// (DirStore cannot know it without reading the file).
+type Opener interface {
+ Open(ctx context.Context, key string) (io.ReadCloser, string, error)
+}
+
+var (
+ _ Opener = (*DirStore)(nil)
+ _ Opener = (*S3Store)(nil)
+ _ Opener = prefixStore{}
+)
+
+// OpenObject streams key from s, buffering through Get when s cannot stream.
+func OpenObject(ctx context.Context, s Store, key string) (io.ReadCloser, string, error) {
+ if o, ok := s.(Opener); ok {
+ return o.Open(ctx, key)
+ }
+ b, etag, err := s.Get(ctx, key)
+ if err != nil {
+ return nil, "", err
+ }
+ return io.NopCloser(bytes.NewReader(b)), etag, nil
+}
+
// ---- DirStore ---------------------------------------------------------------
// DirStore keeps objects as files under a root directory — the test double
@@ -239,6 +272,21 @@ func (d *DirStore) Get(_ context.Context, key string) ([]byte, string, error) {
return b, etagOf(b), nil
}
+func (d *DirStore) Open(_ context.Context, key string) (io.ReadCloser, string, error) {
+ p, err := d.path(key)
+ if err != nil {
+ return nil, "", err
+ }
+ f, err := os.Open(p)
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ return nil, "", fmt.Errorf("dirstore: %s: %w", key, os.ErrNotExist)
+ }
+ return nil, "", err
+ }
+ return f, "", nil
+}
+
func (d *DirStore) Put(_ context.Context, key string, body []byte, ifMatch string) (string, error) {
p, err := d.path(key)
if err != nil {
@@ -505,6 +553,17 @@ func (s *S3Store) Get(ctx context.Context, key string) ([]byte, string, error) {
return b, aws.ToString(out.ETag), nil
}
+func (s *S3Store) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
+ out, err := s.Client.GetObject(ctx, &awss3.GetObjectInput{Bucket: &s.Bucket, Key: &key})
+ if err != nil {
+ if s3NotFound(err) {
+ return nil, "", fmt.Errorf("s3store: %s: %w", key, os.ErrNotExist)
+ }
+ return nil, "", err
+ }
+ return out.Body, aws.ToString(out.ETag), nil
+}
+
func (s *S3Store) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
in := &awss3.PutObjectInput{
Bucket: &s.Bucket,