Move asset reconciliation into shared delivery
This commit is contained in:
@@ -1,6 +1,6 @@
|
|||||||
# Implementation plan: neutral forge delivery and credential preflight
|
# Implementation plan: neutral forge delivery and credential preflight
|
||||||
|
|
||||||
Status: implementation in progress; Stages 1–6 are complete and Stages 7–10
|
Status: implementation in progress; Stages 1–7 are complete and Stages 8–10
|
||||||
are pending.
|
are pending.
|
||||||
|
|
||||||
## Purpose and execution
|
## Purpose and execution
|
||||||
@@ -394,7 +394,7 @@ backend methods.
|
|||||||
|
|
||||||
## Stage 7: Move remote asset comparison and upload recovery
|
## Stage 7: Move remote asset comparison and upload recovery
|
||||||
|
|
||||||
Status: pending.
|
Status: complete.
|
||||||
|
|
||||||
1. Move inspections, asset states/identities, upload aggregation, ordering,
|
1. Move inspections, asset states/identities, upload aggregation, ordering,
|
||||||
and recovery into `internal/delivery`. Use `ListAssets` and `OpenAsset` for
|
and recovery into `internal/delivery`. Use `ListAssets` and `OpenAsset` for
|
||||||
|
|||||||
262
internal/delivery/remote_assets.go
Normal file
262
internal/delivery/remote_assets.go
Normal file
@@ -0,0 +1,262 @@
|
|||||||
|
package delivery
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"crypto/sha256"
|
||||||
|
"encoding/hex"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"os"
|
||||||
|
"sort"
|
||||||
|
|
||||||
|
"go.maximumdirect.net/releasekit/internal/artifact"
|
||||||
|
"go.maximumdirect.net/releasekit/internal/forge"
|
||||||
|
)
|
||||||
|
|
||||||
|
type AssetState string
|
||||||
|
|
||||||
|
const (
|
||||||
|
AssetMatching AssetState = "matching"
|
||||||
|
AssetMissing AssetState = "missing"
|
||||||
|
AssetConflicting AssetState = "conflicting"
|
||||||
|
AssetUnverifiable AssetState = "unverifiable"
|
||||||
|
)
|
||||||
|
|
||||||
|
type AssetIdentity struct {
|
||||||
|
Filename string
|
||||||
|
SHA256 string
|
||||||
|
}
|
||||||
|
|
||||||
|
type AssetInspection struct {
|
||||||
|
Expected artifact.ExpectedAsset
|
||||||
|
State AssetState
|
||||||
|
Existing []forge.Asset
|
||||||
|
Err error
|
||||||
|
}
|
||||||
|
|
||||||
|
type Inspection struct {
|
||||||
|
Assets []AssetInspection
|
||||||
|
Unrelated []forge.Asset
|
||||||
|
}
|
||||||
|
|
||||||
|
type UploadResult struct {
|
||||||
|
State State
|
||||||
|
Matching []AssetIdentity
|
||||||
|
Uploaded []AssetIdentity
|
||||||
|
Pending []AssetIdentity
|
||||||
|
Conflicting []AssetIdentity
|
||||||
|
Uncertain []AssetIdentity
|
||||||
|
}
|
||||||
|
|
||||||
|
func inspectRemoteAssets(ctx context.Context, backend forge.Backend, releaseHandle any, expected []artifact.ExpectedAsset) (Inspection, error) {
|
||||||
|
if err := artifact.ValidateExpectedAssets(expected); err != nil {
|
||||||
|
return Inspection{}, &forge.Failure{Kind: forge.FailureInvalid, Err: err}
|
||||||
|
}
|
||||||
|
listed, err := backend.ListAssets(ctx, releaseHandle)
|
||||||
|
if err != nil {
|
||||||
|
inspection := Inspection{Assets: make([]AssetInspection, 0, len(expected))}
|
||||||
|
for _, item := range expected {
|
||||||
|
inspection.Assets = append(inspection.Assets, AssetInspection{Expected: item, State: AssetUnverifiable, Err: err})
|
||||||
|
}
|
||||||
|
return inspection, err
|
||||||
|
}
|
||||||
|
|
||||||
|
byName := make(map[string][]forge.Asset, len(listed))
|
||||||
|
for _, item := range listed {
|
||||||
|
byName[item.Filename] = append(byName[item.Filename], item)
|
||||||
|
}
|
||||||
|
inspection := Inspection{Assets: make([]AssetInspection, 0, len(expected))}
|
||||||
|
declared := make(map[string]struct{}, len(expected))
|
||||||
|
for _, item := range expected {
|
||||||
|
declared[item.Filename] = struct{}{}
|
||||||
|
matches := byName[item.Filename]
|
||||||
|
observation := AssetInspection{Expected: item, Existing: matches}
|
||||||
|
switch len(matches) {
|
||||||
|
case 0:
|
||||||
|
observation.State = AssetMissing
|
||||||
|
case 1:
|
||||||
|
digest, err := remoteAssetDigest(ctx, backend, matches[0].Handle)
|
||||||
|
if err != nil {
|
||||||
|
observation.State = AssetUnverifiable
|
||||||
|
observation.Err = err
|
||||||
|
} else if digest == item.SHA256 {
|
||||||
|
observation.State = AssetMatching
|
||||||
|
} else {
|
||||||
|
observation.State = AssetConflicting
|
||||||
|
}
|
||||||
|
default:
|
||||||
|
observation.State = AssetConflicting
|
||||||
|
}
|
||||||
|
inspection.Assets = append(inspection.Assets, observation)
|
||||||
|
}
|
||||||
|
for _, item := range listed {
|
||||||
|
if _, ok := declared[item.Filename]; !ok {
|
||||||
|
inspection.Unrelated = append(inspection.Unrelated, item)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return inspection, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func remoteAssetDigest(ctx context.Context, backend forge.Backend, handle any) (string, error) {
|
||||||
|
stream, err := backend.OpenAsset(ctx, handle)
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
hasher := sha256.New()
|
||||||
|
_, readErr := io.Copy(hasher, stream)
|
||||||
|
closeErr := stream.Close()
|
||||||
|
if err := errors.Join(readErr, closeErr); err != nil {
|
||||||
|
return "", &forge.Failure{Kind: forge.FailureUnverifiable, Err: err}
|
||||||
|
}
|
||||||
|
return hex.EncodeToString(hasher.Sum(nil)), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func uploadMissingAssets(ctx context.Context, backend forge.Backend, releaseHandle any, inspection Inspection) (UploadResult, error) {
|
||||||
|
expected := orderedExpected(inspection.Assets)
|
||||||
|
if err := artifact.ValidateExpectedAssets(expected); err != nil {
|
||||||
|
return UploadResult{State: Partial}, &forge.Failure{Kind: forge.FailureInvalid, Err: err}
|
||||||
|
}
|
||||||
|
result := UploadResult{State: Partial}
|
||||||
|
preflightConflict := false
|
||||||
|
preflightUnverifiable := false
|
||||||
|
missing := make(map[string]artifact.ExpectedAsset, len(expected))
|
||||||
|
for _, observation := range inspection.Assets {
|
||||||
|
item := identity(observation.Expected)
|
||||||
|
switch observation.State {
|
||||||
|
case AssetMatching:
|
||||||
|
result.Matching = append(result.Matching, item)
|
||||||
|
case AssetMissing:
|
||||||
|
result.Pending = append(result.Pending, item)
|
||||||
|
missing[observation.Expected.Filename] = observation.Expected
|
||||||
|
case AssetConflicting:
|
||||||
|
preflightConflict = true
|
||||||
|
result.Conflicting = append(result.Conflicting, item)
|
||||||
|
case AssetUnverifiable:
|
||||||
|
preflightUnverifiable = true
|
||||||
|
result.Uncertain = append(result.Uncertain, item)
|
||||||
|
default:
|
||||||
|
return result, &forge.Failure{Kind: forge.FailureInvalid, Err: fmt.Errorf("asset %q has no inspection result", observation.Expected.Filename)}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if preflightConflict {
|
||||||
|
result.State = Conflict
|
||||||
|
return result, &forge.Failure{Kind: forge.FailureConflict, Err: errors.New("one or more existing attachments conflict")}
|
||||||
|
}
|
||||||
|
if preflightUnverifiable {
|
||||||
|
result.State = NotAttempted
|
||||||
|
return result, &forge.Failure{Kind: forge.FailureUnverifiable, Err: errors.New("one or more existing attachments could not be verified")}
|
||||||
|
}
|
||||||
|
if len(missing) == 0 {
|
||||||
|
result.State = Complete
|
||||||
|
return result, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, item := range orderMissing(expected) {
|
||||||
|
if _, needsUpload := missing[item.Filename]; !needsUpload {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
content, err := assetContent(item)
|
||||||
|
if err != nil {
|
||||||
|
return result, &forge.Failure{Kind: forge.FailureInvalid, Err: fmt.Errorf("open local release asset %q: %w", item.Filename, err)}
|
||||||
|
}
|
||||||
|
evidence, uploadErr := backend.UploadAsset(ctx, releaseHandle, item.Filename, content)
|
||||||
|
closeErr := content.Close()
|
||||||
|
if uploadErr == nil {
|
||||||
|
result.Uploaded = append(result.Uploaded, identity(item))
|
||||||
|
result.Pending = removeIdentity(result.Pending, item.Filename)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
operationErr := errors.Join(uploadErr, closeErr)
|
||||||
|
if evidence.Disposition != forge.WriteConflict && evidence.Disposition != forge.WriteUncertain {
|
||||||
|
return result, operationErr
|
||||||
|
}
|
||||||
|
if err := reconcileAssetUpload(ctx, backend, releaseHandle, item, evidence, operationErr, &result); err != nil {
|
||||||
|
return result, err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(result.Pending) == 0 {
|
||||||
|
result.State = Complete
|
||||||
|
}
|
||||||
|
return result, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func reconcileAssetUpload(ctx context.Context, backend forge.Backend, releaseHandle any, item artifact.ExpectedAsset, evidence forge.WriteEvidence, uploadErr error, result *UploadResult) error {
|
||||||
|
inspection, inspectErr := inspectRemoteAssets(ctx, backend, releaseHandle, []artifact.ExpectedAsset{item})
|
||||||
|
if inspectErr != nil || len(inspection.Assets) != 1 {
|
||||||
|
result.State = Unknown
|
||||||
|
result.Pending = removeIdentity(result.Pending, item.Filename)
|
||||||
|
result.Uncertain = append(result.Uncertain, identity(item))
|
||||||
|
return &forge.Failure{Kind: forge.FailureUnknown, Err: errors.Join(uploadErr, inspectErr)}
|
||||||
|
}
|
||||||
|
observation := inspection.Assets[0]
|
||||||
|
switch observation.State {
|
||||||
|
case AssetMatching:
|
||||||
|
result.Matching = append(result.Matching, identity(item))
|
||||||
|
result.Pending = removeIdentity(result.Pending, item.Filename)
|
||||||
|
return nil
|
||||||
|
case AssetConflicting:
|
||||||
|
result.State = Conflict
|
||||||
|
result.Pending = removeIdentity(result.Pending, item.Filename)
|
||||||
|
result.Conflicting = append(result.Conflicting, identity(item))
|
||||||
|
return &forge.Failure{Kind: forge.FailureConflict, Err: uploadErr}
|
||||||
|
case AssetMissing:
|
||||||
|
result.Pending = removeIdentity(result.Pending, item.Filename)
|
||||||
|
if evidence.Disposition == forge.WriteConflict {
|
||||||
|
result.State = Conflict
|
||||||
|
result.Conflicting = append(result.Conflicting, identity(item))
|
||||||
|
return &forge.Failure{Kind: forge.FailureConflict, Err: uploadErr}
|
||||||
|
}
|
||||||
|
result.State = Unknown
|
||||||
|
result.Uncertain = append(result.Uncertain, identity(item))
|
||||||
|
return &forge.Failure{Kind: forge.FailureUnknown, Err: uploadErr}
|
||||||
|
default:
|
||||||
|
result.State = Unknown
|
||||||
|
result.Pending = removeIdentity(result.Pending, item.Filename)
|
||||||
|
result.Uncertain = append(result.Uncertain, identity(item))
|
||||||
|
return &forge.Failure{Kind: forge.FailureUnknown, Err: errors.Join(uploadErr, observation.Err)}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func orderMissing(expected []artifact.ExpectedAsset) []artifact.ExpectedAsset {
|
||||||
|
result := append([]artifact.ExpectedAsset(nil), expected...)
|
||||||
|
sort.Slice(result, func(i, j int) bool {
|
||||||
|
if result[i].Filename == artifact.ManifestFilename {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
if result[j].Filename == artifact.ManifestFilename {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
return result[i].Filename < result[j].Filename
|
||||||
|
})
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
func assetContent(asset artifact.ExpectedAsset) (io.ReadCloser, error) {
|
||||||
|
if len(asset.Content) > 0 {
|
||||||
|
return io.NopCloser(bytes.NewReader(asset.Content)), nil
|
||||||
|
}
|
||||||
|
return os.Open(asset.Path)
|
||||||
|
}
|
||||||
|
|
||||||
|
func orderedExpected(observations []AssetInspection) []artifact.ExpectedAsset {
|
||||||
|
result := make([]artifact.ExpectedAsset, 0, len(observations))
|
||||||
|
for _, observation := range observations {
|
||||||
|
result = append(result, observation.Expected)
|
||||||
|
}
|
||||||
|
return orderMissing(result)
|
||||||
|
}
|
||||||
|
|
||||||
|
func identity(asset artifact.ExpectedAsset) AssetIdentity {
|
||||||
|
return AssetIdentity{Filename: asset.Filename, SHA256: asset.SHA256}
|
||||||
|
}
|
||||||
|
|
||||||
|
func removeIdentity(identities []AssetIdentity, filename string) []AssetIdentity {
|
||||||
|
for index, item := range identities {
|
||||||
|
if item.Filename == filename {
|
||||||
|
return append(identities[:index], identities[index+1:]...)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return identities
|
||||||
|
}
|
||||||
394
internal/delivery/remote_assets_test.go
Normal file
394
internal/delivery/remote_assets_test.go
Normal file
@@ -0,0 +1,394 @@
|
|||||||
|
package delivery
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"io"
|
||||||
|
"os"
|
||||||
|
"reflect"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"go.maximumdirect.net/releasekit/internal/artifact"
|
||||||
|
"go.maximumdirect.net/releasekit/internal/forge"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestInspectRemoteAssetsComparesContentAndPreservesUnrelatedAssets(t *testing.T) {
|
||||||
|
backend := &assetBackend{
|
||||||
|
lists: []assetListResult{{assets: []forge.Asset{
|
||||||
|
{Filename: "same.bin", Size: 4, Handle: "same"},
|
||||||
|
{Filename: "different.bin", Size: 4, Handle: "different"},
|
||||||
|
{Filename: "duplicate.bin", Size: 4, Handle: "first"},
|
||||||
|
{Filename: "duplicate.bin", Size: 4, Handle: "second"},
|
||||||
|
{Filename: "unrelated.txt", Size: 3, Handle: "unrelated"},
|
||||||
|
}}},
|
||||||
|
content: map[string][]byte{
|
||||||
|
"same": []byte("same"),
|
||||||
|
"different": []byte("diff"),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
expected := []artifact.ExpectedAsset{
|
||||||
|
expectedAsset("same.bin", []byte("same")),
|
||||||
|
expectedAsset("different.bin", []byte("else")),
|
||||||
|
expectedAsset("duplicate.bin", []byte("dupe")),
|
||||||
|
expectedAsset("missing.bin", []byte("gone")),
|
||||||
|
}
|
||||||
|
|
||||||
|
inspection, err := inspectRemoteAssets(context.Background(), backend, "release", expected)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
wantStates := []AssetState{AssetMatching, AssetConflicting, AssetConflicting, AssetMissing}
|
||||||
|
for i, observation := range inspection.Assets {
|
||||||
|
if observation.State != wantStates[i] {
|
||||||
|
t.Errorf("asset %q state = %s, want %s", observation.Expected.Filename, observation.State, wantStates[i])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(inspection.Unrelated) != 1 || inspection.Unrelated[0].Filename != "unrelated.txt" {
|
||||||
|
t.Fatalf("unrelated assets = %#v", inspection.Unrelated)
|
||||||
|
}
|
||||||
|
if len(backend.opened) != 2 || !backend.opened[0].closed || !backend.opened[1].closed {
|
||||||
|
t.Fatalf("opened asset streams = %#v", backend.opened)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestInspectRemoteAssetsListingFailureDoesNotBecomeUnknownWrite(t *testing.T) {
|
||||||
|
assets := []artifact.ExpectedAsset{expectedAsset("a.bin", []byte("a")), expectedAsset("b.bin", []byte("b"))}
|
||||||
|
listErr := &forge.Failure{Kind: forge.FailureTransport, Err: errors.New("service unavailable")}
|
||||||
|
backend := &assetBackend{lists: []assetListResult{{err: listErr}}}
|
||||||
|
inspection, err := inspectRemoteAssets(context.Background(), backend, "release", assets)
|
||||||
|
if !errors.Is(err, listErr) || len(inspection.Assets) != len(assets) {
|
||||||
|
t.Fatalf("inspection = %#v, error %v", inspection, err)
|
||||||
|
}
|
||||||
|
for _, observation := range inspection.Assets {
|
||||||
|
if observation.State != AssetUnverifiable || !errors.Is(observation.Err, listErr) {
|
||||||
|
t.Fatalf("listing failure observation = %#v", observation)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
result, err := uploadMissingAssets(context.Background(), backend, "release", inspection)
|
||||||
|
if !isForgeFailure(err, forge.FailureUnverifiable) || result.State != NotAttempted || len(result.Uncertain) != len(assets) || len(backend.uploadNames) != 0 {
|
||||||
|
t.Fatalf("uploadMissingAssets() = %#v, error %v, uploads %v", result, err, backend.uploadNames)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestInspectRemoteAssetsKeepsMatchingManifestWhenBinariesAreMissing(t *testing.T) {
|
||||||
|
binary := expectedAsset("app-linux-amd64", []byte("binary"))
|
||||||
|
manifest, err := artifact.BuildManifest([]artifact.ExpectedAsset{binary})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
backend := &assetBackend{
|
||||||
|
lists: []assetListResult{{assets: []forge.Asset{{Filename: manifest.Filename, Size: int64(len(manifest.Content)), Handle: "manifest"}}}},
|
||||||
|
content: map[string][]byte{"manifest": manifest.Content},
|
||||||
|
}
|
||||||
|
inspection, err := inspectRemoteAssets(context.Background(), backend, "release", []artifact.ExpectedAsset{binary, manifest})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if inspection.Assets[0].State != AssetMissing || inspection.Assets[1].State != AssetMatching {
|
||||||
|
t.Fatalf("manifest inspection = %#v", inspection.Assets)
|
||||||
|
}
|
||||||
|
result, err := uploadMissingAssets(context.Background(), backend, "release", inspection)
|
||||||
|
if err != nil || result.State != Complete || !reflect.DeepEqual(result.Uploaded, []AssetIdentity{identity(binary)}) {
|
||||||
|
t.Fatalf("uploadMissingAssets() = %#v, error %v", result, err)
|
||||||
|
}
|
||||||
|
if !reflect.DeepEqual(backend.uploadNames, []string{binary.Filename}) {
|
||||||
|
t.Fatalf("uploaded names = %v", backend.uploadNames)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestInspectRemoteAssetsRejectsSubsetOrDifferentManifestContent(t *testing.T) {
|
||||||
|
binaryA := expectedAsset("a", []byte("a"))
|
||||||
|
binaryB := expectedAsset("b", []byte("b"))
|
||||||
|
manifest, err := artifact.BuildManifest([]artifact.ExpectedAsset{binaryA, binaryB})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
for _, test := range []struct {
|
||||||
|
name string
|
||||||
|
content []byte
|
||||||
|
}{
|
||||||
|
{name: "subset", content: []byte("a a\n")},
|
||||||
|
{name: "different", content: []byte("different manifest\n")},
|
||||||
|
} {
|
||||||
|
t.Run(test.name, func(t *testing.T) {
|
||||||
|
backend := &assetBackend{
|
||||||
|
lists: []assetListResult{{assets: []forge.Asset{{Filename: artifact.ManifestFilename, Handle: "manifest"}}}},
|
||||||
|
content: map[string][]byte{"manifest": test.content},
|
||||||
|
}
|
||||||
|
inspection, err := inspectRemoteAssets(context.Background(), backend, "release", []artifact.ExpectedAsset{binaryA, binaryB, manifest})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if inspection.Assets[2].State != AssetConflicting {
|
||||||
|
t.Fatalf("manifest state = %s, want conflicting", inspection.Assets[2].State)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUploadMissingAssetsPreflightsAndOrdersWrites(t *testing.T) {
|
||||||
|
assets := []artifact.ExpectedAsset{
|
||||||
|
expectedAsset("z.bin", []byte("z")),
|
||||||
|
expectedAsset(artifact.ManifestFilename, []byte("manifest")),
|
||||||
|
expectedAsset("a.bin", []byte("a")),
|
||||||
|
}
|
||||||
|
inspection := inspectionFor(assets, map[string]AssetState{
|
||||||
|
"z.bin": AssetMissing,
|
||||||
|
artifact.ManifestFilename: AssetMissing,
|
||||||
|
"a.bin": AssetMissing,
|
||||||
|
})
|
||||||
|
backend := &assetBackend{uploads: []assetUploadResult{
|
||||||
|
{evidence: forge.WriteEvidence{Attempted: true, Disposition: forge.WriteAccepted}},
|
||||||
|
{evidence: forge.WriteEvidence{Attempted: true, Disposition: forge.WriteAccepted}},
|
||||||
|
{evidence: forge.WriteEvidence{Attempted: true, Disposition: forge.WriteAccepted}},
|
||||||
|
}}
|
||||||
|
result, err := uploadMissingAssets(context.Background(), backend, "release", inspection)
|
||||||
|
if err != nil || result.State != Complete {
|
||||||
|
t.Fatalf("uploadMissingAssets() = %#v, error %v", result, err)
|
||||||
|
}
|
||||||
|
want := []string{"a.bin", "z.bin", artifact.ManifestFilename}
|
||||||
|
if !reflect.DeepEqual(backend.uploadNames, want) {
|
||||||
|
t.Fatalf("upload order = %v, want %v", backend.uploadNames, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUploadMissingAssetsStopsBeforeWritesForConflictOrUnverifiableContent(t *testing.T) {
|
||||||
|
for _, test := range []struct {
|
||||||
|
name string
|
||||||
|
state AssetState
|
||||||
|
want State
|
||||||
|
}{
|
||||||
|
{name: "conflict", state: AssetConflicting, want: Conflict},
|
||||||
|
{name: "unverifiable", state: AssetUnverifiable, want: NotAttempted},
|
||||||
|
} {
|
||||||
|
t.Run(test.name, func(t *testing.T) {
|
||||||
|
assets := []artifact.ExpectedAsset{expectedAsset("existing", []byte("old")), expectedAsset("missing", []byte("new"))}
|
||||||
|
inspection := inspectionFor(assets, map[string]AssetState{"existing": test.state, "missing": AssetMissing})
|
||||||
|
result, err := uploadMissingAssets(context.Background(), &assetBackend{}, "release", inspection)
|
||||||
|
if err == nil || result.State != test.want || len(result.Pending) != 1 || len(result.Conflicting)+len(result.Uncertain) != 1 {
|
||||||
|
t.Fatalf("uploadMissingAssets() = %#v, error %v", result, err)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUploadMissingAssetsReconcilesOnceAndKeepsLaterAssetsPending(t *testing.T) {
|
||||||
|
assets := []artifact.ExpectedAsset{
|
||||||
|
expectedAsset("a.bin", []byte("a")),
|
||||||
|
expectedAsset("b.bin", []byte("b")),
|
||||||
|
expectedAsset("c.bin", []byte("c")),
|
||||||
|
}
|
||||||
|
for _, test := range []struct {
|
||||||
|
name string
|
||||||
|
disposition forge.WriteDisposition
|
||||||
|
followupAssets []forge.Asset
|
||||||
|
followupData map[string][]byte
|
||||||
|
openErrors map[string]error
|
||||||
|
followupErr error
|
||||||
|
wantState State
|
||||||
|
wantCategory string
|
||||||
|
wantContinue bool
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "matching bytes resolve uncertain upload", disposition: forge.WriteUncertain,
|
||||||
|
followupAssets: []forge.Asset{{Filename: "a.bin", Handle: "a"}}, followupData: map[string][]byte{"a": []byte("a")},
|
||||||
|
wantState: Complete, wantCategory: "matching", wantContinue: true,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "different bytes conflict", disposition: forge.WriteUncertain,
|
||||||
|
followupAssets: []forge.Asset{{Filename: "a.bin", Handle: "a"}}, followupData: map[string][]byte{"a": []byte("x")},
|
||||||
|
wantState: Conflict, wantCategory: "conflicting",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "missing after conflict remains conflict", disposition: forge.WriteConflict,
|
||||||
|
wantState: Conflict, wantCategory: "conflicting",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "missing after uncertain upload is unknown", disposition: forge.WriteUncertain,
|
||||||
|
wantState: Unknown, wantCategory: "uncertain",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "unreadable follow-up is unknown", disposition: forge.WriteUncertain,
|
||||||
|
followupAssets: []forge.Asset{{Filename: "a.bin", Handle: "a"}}, openErrors: map[string]error{"a": errors.New("download failed")},
|
||||||
|
wantState: Unknown, wantCategory: "uncertain",
|
||||||
|
},
|
||||||
|
} {
|
||||||
|
t.Run(test.name, func(t *testing.T) {
|
||||||
|
first := assetListResult{assets: []forge.Asset{}}
|
||||||
|
followup := assetListResult{assets: test.followupAssets, err: test.followupErr}
|
||||||
|
backend := &assetBackend{
|
||||||
|
lists: []assetListResult{first, followup},
|
||||||
|
content: test.followupData,
|
||||||
|
openErrors: test.openErrors,
|
||||||
|
uploads: []assetUploadResult{{
|
||||||
|
evidence: forge.WriteEvidence{Attempted: true, Disposition: test.disposition},
|
||||||
|
err: &forge.Failure{Kind: failureKindForDisposition(test.disposition), Err: errors.New("upload outcome unresolved")},
|
||||||
|
}},
|
||||||
|
}
|
||||||
|
inspection, err := inspectRemoteAssets(context.Background(), backend, "release", assets)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
result, uploadErr := uploadMissingAssets(context.Background(), backend, "release", inspection)
|
||||||
|
if result.State != test.wantState || backend.listCalls != 2 {
|
||||||
|
t.Fatalf("result = %#v, error %v, list calls %d, uploads %v", result, uploadErr, backend.listCalls, backend.uploadNames)
|
||||||
|
}
|
||||||
|
if test.wantContinue {
|
||||||
|
if uploadErr != nil || !reflect.DeepEqual(backend.uploadNames, []string{"a.bin", "b.bin", "c.bin"}) || len(result.Pending) != 0 {
|
||||||
|
t.Fatalf("continued result = %#v, error %v, uploads %v", result, uploadErr, backend.uploadNames)
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
if uploadErr == nil || !reflect.DeepEqual(backend.uploadNames, []string{"a.bin"}) {
|
||||||
|
t.Fatalf("stopped result = %#v, error %v, uploads %v", result, uploadErr, backend.uploadNames)
|
||||||
|
}
|
||||||
|
if !reflect.DeepEqual(result.Pending, []AssetIdentity{identity(assets[1]), identity(assets[2])}) {
|
||||||
|
t.Fatalf("pending = %#v", result.Pending)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
var category []AssetIdentity
|
||||||
|
switch test.wantCategory {
|
||||||
|
case "matching":
|
||||||
|
category = result.Matching
|
||||||
|
case "conflicting":
|
||||||
|
category = result.Conflicting
|
||||||
|
case "uncertain":
|
||||||
|
category = result.Uncertain
|
||||||
|
}
|
||||||
|
if len(category) != 1 || category[0].Filename != "a.bin" {
|
||||||
|
t.Fatalf("%s assets = %#v", test.wantCategory, category)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUploadMissingAssetsPreservesAcknowledgedProgressOnDefinitiveFailure(t *testing.T) {
|
||||||
|
assets := []artifact.ExpectedAsset{expectedAsset("a.bin", []byte("a")), expectedAsset("b.bin", []byte("b")), expectedAsset("c.bin", []byte("c"))}
|
||||||
|
backend := &assetBackend{uploads: []assetUploadResult{
|
||||||
|
{evidence: forge.WriteEvidence{Attempted: true, Disposition: forge.WriteAccepted}},
|
||||||
|
{evidence: forge.WriteEvidence{Attempted: true, Disposition: forge.WriteRejected}, err: &forge.Failure{Kind: forge.FailureRejected, Err: errors.New("denied")}},
|
||||||
|
}}
|
||||||
|
result, err := uploadMissingAssets(context.Background(), backend, "release", inspectionFor(assets, map[string]AssetState{
|
||||||
|
"a.bin": AssetMissing, "b.bin": AssetMissing, "c.bin": AssetMissing,
|
||||||
|
}))
|
||||||
|
if err == nil || result.State != Partial || !reflect.DeepEqual(result.Uploaded, []AssetIdentity{identity(assets[0])}) || !reflect.DeepEqual(result.Pending, []AssetIdentity{identity(assets[1]), identity(assets[2])}) {
|
||||||
|
t.Fatalf("uploadMissingAssets() = %#v, error %v", result, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUploadMissingAssetsBorrowsReaderAndClosesItAfterCall(t *testing.T) {
|
||||||
|
filePath := t.TempDir() + "/asset.bin"
|
||||||
|
if err := os.WriteFile(filePath, []byte("asset"), 0o600); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
asset := artifact.ExpectedAsset{Filename: "asset.bin", SHA256: assetDigest([]byte("asset")), Size: 5, Path: filePath}
|
||||||
|
backend := &assetBackend{uploads: []assetUploadResult{{evidence: forge.WriteEvidence{Attempted: true, Disposition: forge.WriteAccepted}}}}
|
||||||
|
result, err := uploadMissingAssets(context.Background(), backend, "release", inspectionFor([]artifact.ExpectedAsset{asset}, map[string]AssetState{"asset.bin": AssetMissing}))
|
||||||
|
if err != nil || result.State != Complete || !backend.readerWasOpen {
|
||||||
|
t.Fatalf("uploadMissingAssets() = %#v, error %v, reader open during call %v", result, err, backend.readerWasOpen)
|
||||||
|
}
|
||||||
|
if backend.borrowed == nil {
|
||||||
|
t.Fatal("backend did not receive the file reader")
|
||||||
|
}
|
||||||
|
if _, err := backend.borrowed.Stat(); !errors.Is(err, os.ErrClosed) {
|
||||||
|
t.Fatalf("borrowed file stat error = %v, want closed file", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type assetListResult struct {
|
||||||
|
assets []forge.Asset
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
|
type assetUploadResult struct {
|
||||||
|
evidence forge.WriteEvidence
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
|
||||||
|
type assetBackend struct {
|
||||||
|
pageBackend
|
||||||
|
lists []assetListResult
|
||||||
|
listCalls int
|
||||||
|
content map[string][]byte
|
||||||
|
openErrors map[string]error
|
||||||
|
opened []*trackedAssetStream
|
||||||
|
uploads []assetUploadResult
|
||||||
|
uploadNames []string
|
||||||
|
uploadContent [][]byte
|
||||||
|
borrowed *os.File
|
||||||
|
readerWasOpen bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (backend *assetBackend) ListAssets(context.Context, any) ([]forge.Asset, error) {
|
||||||
|
index := backend.listCalls
|
||||||
|
backend.listCalls++
|
||||||
|
if index >= len(backend.lists) {
|
||||||
|
return nil, errors.New("unexpected asset listing")
|
||||||
|
}
|
||||||
|
result := backend.lists[index]
|
||||||
|
return result.assets, result.err
|
||||||
|
}
|
||||||
|
|
||||||
|
func (backend *assetBackend) OpenAsset(_ context.Context, handle any) (io.ReadCloser, error) {
|
||||||
|
key, ok := handle.(string)
|
||||||
|
if !ok {
|
||||||
|
return nil, errors.New("unexpected asset handle")
|
||||||
|
}
|
||||||
|
if err := backend.openErrors[key]; err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
stream := &trackedAssetStream{Reader: bytes.NewReader(backend.content[key])}
|
||||||
|
backend.opened = append(backend.opened, stream)
|
||||||
|
return stream, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (backend *assetBackend) UploadAsset(_ context.Context, _ any, filename string, content io.Reader) (forge.WriteEvidence, error) {
|
||||||
|
backend.uploadNames = append(backend.uploadNames, filename)
|
||||||
|
if file, ok := content.(*os.File); ok {
|
||||||
|
backend.borrowed = file
|
||||||
|
}
|
||||||
|
data, err := io.ReadAll(content)
|
||||||
|
backend.uploadContent = append(backend.uploadContent, data)
|
||||||
|
if backend.borrowed != nil {
|
||||||
|
_, statErr := backend.borrowed.Stat()
|
||||||
|
backend.readerWasOpen = statErr == nil
|
||||||
|
}
|
||||||
|
index := len(backend.uploadNames) - 1
|
||||||
|
if err != nil {
|
||||||
|
return forge.WriteEvidence{Attempted: true, Disposition: forge.WriteUncertain}, err
|
||||||
|
}
|
||||||
|
if index >= len(backend.uploads) {
|
||||||
|
return forge.WriteEvidence{Attempted: true, Disposition: forge.WriteAccepted}, nil
|
||||||
|
}
|
||||||
|
result := backend.uploads[index]
|
||||||
|
return result.evidence, result.err
|
||||||
|
}
|
||||||
|
|
||||||
|
type trackedAssetStream struct {
|
||||||
|
*bytes.Reader
|
||||||
|
closed bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (stream *trackedAssetStream) Close() error {
|
||||||
|
stream.closed = true
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func expectedAsset(filename string, content []byte) artifact.ExpectedAsset {
|
||||||
|
return artifact.ExpectedAsset{Filename: filename, SHA256: assetDigest(content), Size: int64(len(content)), Content: content}
|
||||||
|
}
|
||||||
|
|
||||||
|
func inspectionFor(assets []artifact.ExpectedAsset, states map[string]AssetState) Inspection {
|
||||||
|
inspection := Inspection{Assets: make([]AssetInspection, 0, len(assets))}
|
||||||
|
for _, item := range assets {
|
||||||
|
inspection.Assets = append(inspection.Assets, AssetInspection{Expected: item, State: states[item.Filename]})
|
||||||
|
}
|
||||||
|
return inspection
|
||||||
|
}
|
||||||
|
|
||||||
|
func failureKindForDisposition(disposition forge.WriteDisposition) forge.FailureKind {
|
||||||
|
if disposition == forge.WriteConflict {
|
||||||
|
return forge.FailureConflict
|
||||||
|
}
|
||||||
|
return forge.FailureTransport
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user