| 1 | package release |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "crypto/sha256" |
| 7 | "encoding/hex" |
| 8 | "errors" |
| 9 | "io" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "runtime" |
| 13 | "strings" |
| 14 | "testing" |
| 15 | ) |
| 16 | |
| 17 | // wholeArtifactRefuser fails the test if anything reads a release artifact |
| 18 | // whole through Get. Manifests and pointers are small and still come through |
| 19 | // Get; a build has to be streamed through Open (#269). |
| 20 | type wholeArtifactRefuser struct { |
| 21 | Store |
| 22 | t *testing.T |
| 23 | } |
| 24 | |
| 25 | func (w wholeArtifactRefuser) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 26 | if strings.Contains(key, "/releases/") && !strings.HasSuffix(key, "manifest.json") { |
| 27 | w.t.Errorf("artifact %s was read whole into memory; staging must stream it", key) |
| 28 | } |
| 29 | return w.Store.Get(ctx, key) |
| 30 | } |
| 31 | |
| 32 | func (w wholeArtifactRefuser) Open(ctx context.Context, key string) (io.ReadCloser, string, error) { |
| 33 | return OpenObject(ctx, w.Store, key) |
| 34 | } |
| 35 | |
| 36 | func TestStagingStreamsTheArtifact(t *testing.T) { |
| 37 | dir := NewDirStore(t.TempDir()) |
| 38 | ctx := context.Background() |
| 39 | opt := t.TempDir() |
| 40 | body := []byte("the whole build") |
| 41 | shipRelease(t, dir, "bab", "km", "v1", adoptPlatform, body) |
| 42 | |
| 43 | if _, err := ensureStaged(ctx, wholeArtifactRefuser{Store: dir, t: t}, opt, "bab", "km", "v1", "h.example.com"); err != nil { |
| 44 | t.Fatalf("ensureStaged: %v", err) |
| 45 | } |
| 46 | got, err := os.ReadFile(VersionPath(opt, "bab", "v1", "km")) |
| 47 | if err != nil || !bytes.Equal(got, body) { |
| 48 | t.Fatalf("staged %q (err %v), want %q", got, err, body) |
| 49 | } |
| 50 | } |
| 51 | |
| 52 | // patternReader yields n bytes of a fixed pattern without holding them. |
| 53 | type patternReader struct{ n int64 } |
| 54 | |
| 55 | func (p *patternReader) Read(b []byte) (int, error) { |
| 56 | if p.n <= 0 { |
| 57 | return 0, io.EOF |
| 58 | } |
| 59 | if int64(len(b)) > p.n { |
| 60 | b = b[:p.n] |
| 61 | } |
| 62 | for i := range b { |
| 63 | b[i] = byte(i * 31) |
| 64 | } |
| 65 | p.n -= int64(len(b)) |
| 66 | return len(b), nil |
| 67 | } |
| 68 | |
| 69 | // The regression itself: a ~99 MB build read whole peaked past the stage |
| 70 | // pass's 256 MB cap. Streamed, staging a build allocates a copy buffer, not |
| 71 | // the build. |
| 72 | func TestStageArtifactFromKeepsMemoryFlat(t *testing.T) { |
| 73 | if testing.Short() { |
| 74 | t.Skip("writes a 96 MB file") |
| 75 | } |
| 76 | const size = 96 << 20 |
| 77 | h := sha256.New() |
| 78 | if _, err := io.Copy(h, &patternReader{n: size}); err != nil { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | want := hex.EncodeToString(h.Sum(nil)) |
| 82 | opt := t.TempDir() |
| 83 | |
| 84 | runtime.GC() |
| 85 | var before, after runtime.MemStats |
| 86 | runtime.ReadMemStats(&before) |
| 87 | err := stageArtifactFrom(opt, "bab", "v1", "big", want, func() (io.ReadCloser, error) { |
| 88 | return io.NopCloser(&patternReader{n: size}), nil |
| 89 | }) |
| 90 | runtime.ReadMemStats(&after) |
| 91 | if err != nil { |
| 92 | t.Fatalf("stageArtifactFrom: %v", err) |
| 93 | } |
| 94 | if grew := after.TotalAlloc - before.TotalAlloc; grew > 8<<20 { |
| 95 | t.Errorf("staging a %d MB build allocated %d MB; it must stream", size>>20, grew>>20) |
| 96 | } |
| 97 | if fi, err := os.Stat(VersionPath(opt, "bab", "v1", "big")); err != nil || fi.Size() != size || fi.Mode().Perm() != 0o755 { |
| 98 | t.Fatalf("staged file: %v, err %v", fi, err) |
| 99 | } |
| 100 | } |
| 101 | |
| 102 | // failingReader yields some bytes, then fails, like a connection dropped |
| 103 | // partway through a download. |
| 104 | type failingReader struct{ sent bool } |
| 105 | |
| 106 | func (f *failingReader) Read(b []byte) (int, error) { |
| 107 | if f.sent { |
| 108 | return 0, errors.New("connection reset") |
| 109 | } |
| 110 | f.sent = true |
| 111 | return copy(b, "half a build"), nil |
| 112 | } |
| 113 | |
| 114 | // Bytes that fail to arrive whole, or arrive wrong, never replace what was |
| 115 | // staged under the name and leave no temp file behind, exactly as verifying |
| 116 | // in memory first did. |
| 117 | func TestStageArtifactFromLeavesTheStagedCopyOnFailure(t *testing.T) { |
| 118 | good := []byte("new build") |
| 119 | sum := sha256.Sum256(good) |
| 120 | for _, tc := range []struct { |
| 121 | name string |
| 122 | want string |
| 123 | r func() io.Reader |
| 124 | mismatch bool |
| 125 | }{ |
| 126 | {"wrong bytes", hex.EncodeToString(sum[:]), func() io.Reader { return strings.NewReader("not the build") }, true}, |
| 127 | {"read fails partway", hex.EncodeToString(sum[:]), func() io.Reader { return &failingReader{} }, false}, |
| 128 | } { |
| 129 | t.Run(tc.name, func(t *testing.T) { |
| 130 | opt := t.TempDir() |
| 131 | if err := StageArtifact(opt, "bab", "v1", "km", []byte("old build")); err != nil { |
| 132 | t.Fatal(err) |
| 133 | } |
| 134 | err := stageArtifactFrom(opt, "bab", "v1", "km", tc.want, func() (io.ReadCloser, error) { |
| 135 | return io.NopCloser(tc.r()), nil |
| 136 | }) |
| 137 | if err == nil || !strings.Contains(err.Error(), "download v1") { |
| 138 | t.Fatalf("err = %v, want a download error", err) |
| 139 | } |
| 140 | if errors.Is(err, errArtifactMismatch) != tc.mismatch { |
| 141 | t.Errorf("errors.Is(err, errArtifactMismatch) = %v, want %v", !tc.mismatch, tc.mismatch) |
| 142 | } |
| 143 | path := VersionPath(opt, "bab", "v1", "km") |
| 144 | if got, _ := os.ReadFile(path); string(got) != "old build" { |
| 145 | t.Errorf("staged copy is %q, want the old build untouched", got) |
| 146 | } |
| 147 | if left, _ := filepath.Glob(filepath.Join(filepath.Dir(path), ".carlos-write-*")); len(left) != 0 { |
| 148 | t.Errorf("temp files left behind: %v", left) |
| 149 | } |
| 150 | }) |
| 151 | } |
| 152 | } |
| 153 | |
| 154 | // A box that cannot write the version store (the sandboxed agent) fails on |
| 155 | // the directory before it opens the build, so it never downloads one only to |
| 156 | // relay the stage to root. |
| 157 | func TestStageArtifactFromChecksTheStoreBeforeOpening(t *testing.T) { |
| 158 | opt := filepath.Join(t.TempDir(), "not-a-dir") |
| 159 | if err := os.WriteFile(opt, nil, 0o644); err != nil { |
| 160 | t.Fatal(err) |
| 161 | } |
| 162 | err := stageArtifactFrom(opt, "bab", "v1", "km", "", func() (io.ReadCloser, error) { |
| 163 | t.Error("the build was opened though the version store cannot be written") |
| 164 | return io.NopCloser(strings.NewReader("")), nil |
| 165 | }) |
| 166 | if err == nil { |
| 167 | t.Fatal("staging into an unwritable store succeeded") |
| 168 | } |
| 169 | } |
| 170 | |
| 171 | // Two writers of one version overlap (Astra, stage-stream-artifact review): |
| 172 | // A has written and hashed the whole build when B, a download that later |
| 173 | // fails, starts writing. With a shared temp file B's truncate-and-write lands |
| 174 | // in A's file before A renames it, and A publishes B's partial bytes under a |
| 175 | // passing hash. Each writer's own temp file keeps A's bytes A's. |
| 176 | func TestStageArtifactFromOverlappingWriters(t *testing.T) { |
| 177 | opt := t.TempDir() |
| 178 | good := []byte("the good build, whole") |
| 179 | sum := sha256.Sum256(good) |
| 180 | want := hex.EncodeToString(sum[:]) |
| 181 | |
| 182 | bWrote, release, bDone := make(chan struct{}), make(chan struct{}), make(chan error, 1) |
| 183 | a := &overlapReader{body: good, atEOF: func() { |
| 184 | go func() { |
| 185 | bDone <- stageArtifactFrom(opt, "bab", "v1", "km", want, func() (io.ReadCloser, error) { |
| 186 | return io.NopCloser(&stallingReader{wrote: bWrote, release: release}), nil |
| 187 | }) |
| 188 | }() |
| 189 | <-bWrote |
| 190 | }} |
| 191 | if err := stageArtifactFrom(opt, "bab", "v1", "km", want, func() (io.ReadCloser, error) { |
| 192 | return io.NopCloser(a), nil |
| 193 | }); err != nil { |
| 194 | t.Fatalf("writer A: %v", err) |
| 195 | } |
| 196 | close(release) |
| 197 | if err := <-bDone; err == nil { |
| 198 | t.Fatal("writer B's failed download reported success") |
| 199 | } |
| 200 | path := VersionPath(opt, "bab", "v1", "km") |
| 201 | if got, _ := os.ReadFile(path); !bytes.Equal(got, good) { |
| 202 | t.Fatalf("staged %q, want writer A's verified bytes %q", got, good) |
| 203 | } |
| 204 | if left, _ := filepath.Glob(filepath.Join(filepath.Dir(path), ".carlos-write-*")); len(left) != 0 { |
| 205 | t.Errorf("temp files left behind: %v", left) |
| 206 | } |
| 207 | } |
| 208 | |
| 209 | // overlapReader yields body, then runs atEOF once before reporting EOF. |
| 210 | type overlapReader struct { |
| 211 | body []byte |
| 212 | atEOF func() |
| 213 | } |
| 214 | |
| 215 | func (o *overlapReader) Read(b []byte) (int, error) { |
| 216 | if len(o.body) > 0 { |
| 217 | n := copy(b, o.body) |
| 218 | o.body = o.body[n:] |
| 219 | return n, nil |
| 220 | } |
| 221 | if o.atEOF != nil { |
| 222 | o.atEOF() |
| 223 | o.atEOF = nil |
| 224 | } |
| 225 | return 0, io.EOF |
| 226 | } |
| 227 | |
| 228 | // stallingReader yields a partial build, signals once it has been written |
| 229 | // (the second Read only comes after the first chunk's write), then waits for |
| 230 | // release and fails. |
| 231 | type stallingReader struct { |
| 232 | n int |
| 233 | wrote chan struct{} |
| 234 | release chan struct{} |
| 235 | } |
| 236 | |
| 237 | func (s *stallingReader) Read(b []byte) (int, error) { |
| 238 | s.n++ |
| 239 | if s.n == 1 { |
| 240 | return copy(b, "partial"), nil |
| 241 | } |
| 242 | close(s.wrote) |
| 243 | <-s.release |
| 244 | return 0, errors.New("connection reset") |
| 245 | } |
| 246 | |
amadan