internal/pkg: in-flight error resolution
Test / Create distribution (push) Successful in 55s
Test / Sandbox (push) Successful in 2m58s
Test / Hakurei (push) Successful in 4m28s
Test / Sandbox (race detector) (push) Successful in 5m55s
Test / Hakurei (race detector) (push) Successful in 7m35s
Test / ShareFS (push) Successful in 8m31s
Test / Flake checks (push) Successful in 1m32s

The DCE is a slow and overcomplicated solution to a simple problem. This change replaces the DCE by resolving errors in-flight.

Signed-off-by: Ophestra <cat@gensokyo.uk>
This commit is contained in:
cat
2026-08-12 16:21:47 +09:00
parent 1f45d44e7f
commit a431f16e6f
4 changed files with 81 additions and 163 deletions
+4
View File
@@ -1196,6 +1196,10 @@ func main() {
if w, ok := err.(interface{ Unwrap() []error }); !ok { if w, ok := err.(interface{ Unwrap() []error }); !ok {
log.Fatal(err) log.Fatal(err)
} else { } else {
if _, ok = w.(pkg.InputError); ok {
log.Fatal(w)
}
errs := w.Unwrap() errs := w.Unwrap()
for i, e := range errs { for i, e := range errs {
if i == len(errs)-1 { if i == len(errs)-1 {
+2 -5
View File
@@ -118,11 +118,8 @@ func TestExec(t *testing.T) {
[]string{"testtool"}, []string{"testtool"},
pkg.MustPath("/proc/nonexistent", false, failingArtifact), pkg.MustPath("/proc/nonexistent", false, failingArtifact),
), nil, nil, pkg.WNew, &pkg.DependencyCureError{ ), nil, nil, pkg.WNew, pkg.InputError{
{ failingArtifact: stub.UniqueError(0xcafe),
A: failingArtifact,
Err: stub.UniqueError(0xcafe),
},
}}, }},
{"invalid paths", pkg.NewExec( {"invalid paths", pkg.NewExec(
+31 -94
View File
@@ -18,7 +18,6 @@ import (
"maps" "maps"
"math" "math"
"os" "os"
"os/signal"
"path/filepath" "path/filepath"
"runtime" "runtime"
"slices" "slices"
@@ -654,10 +653,10 @@ type pendingArtifactDep struct {
// if curing succeeds. // if curing succeeds.
resP *cureRes resP *cureRes
// Address of result error slice populated during [Cache.Cure], dereferenced // Address of result error map populated during [Cache.Cure], dereferenced
// after acquiring errsMu if curing fails. No additional action is taken, // after acquiring errsMu if curing fails. No additional action is taken,
// [Cache] and its caller are responsible for further error handling. // [Cache] and its caller are responsible for further error handling.
errs *DependencyCureError errs InputError
// Address of mutex synchronising access to errs. // Address of mutex synchronising access to errs.
errsMu *sync.Mutex errsMu *sync.Mutex
@@ -1721,109 +1720,43 @@ retry:
goto retry goto retry
} }
// CureError wraps a non-nil error returned attempting to cure an [Artifact]. // An InputError describes inputs of a [FloodArtifact] which had failed to cure.
type CureError struct { type InputError map[Artifact]error
A Artifact
Err error
}
// Unwrap returns the underlying error. // Error returns a user-facing, deterministic text representation of e.
func (e *CureError) Unwrap() error { return e.Err } func (e InputError) Error() string {
ir := NewIR()
// Error returns the error message from the underlying Err. type input struct {
func (e *CureError) Error() string { return e.Err.Error() } a Artifact
id unique.Handle[ID]
// A DependencyCureError wraps errors returned while curing dependencies.
type DependencyCureError []*CureError
// unwrapM recursively expands underlying errors into a caller-supplied map.
func (e *DependencyCureError) unwrapM(
ctx context.Context,
ir *IRCache,
me map[unique.Handle[ID]]*CureError,
) {
for _, err := range *e {
if ctx.Err() != nil {
break
}
id := ir.Ident(err.A)
if _, ok := me[id]; ok {
continue
}
if _e, ok := err.Err.(*DependencyCureError); ok {
_e.unwrapM(ctx, ir, me)
continue
}
me[id] = err
} }
} p := make([]input, 0, len(e))
for a := range e {
// unwrap recursively expands and deduplicates underlying errors. p = append(p, input{a, ir.Ident(a)})
func (e *DependencyCureError) unwrap(
ctx context.Context,
ir *IRCache,
) DependencyCureError {
me := make(map[unique.Handle[ID]]*CureError)
e.unwrapM(ctx, ir, me)
type ent struct {
id unique.Handle[ID]
err *CureError
}
errs := make([]*ent, 0, len(me))
for id, err := range me {
errs = append(errs, &ent{id, err})
} }
var identBuf [2]ID var identBuf [2]ID
slices.SortFunc(errs, func(a, b *ent) int { slices.SortFunc(p, func(a, b input) int {
identBuf[0], identBuf[1] = a.id.Value(), b.id.Value() identBuf[0], identBuf[1] = a.id.Value(), b.id.Value()
return slices.Compare(identBuf[0][:], identBuf[1][:]) return slices.Compare(identBuf[0][:], identBuf[1][:])
}) })
_errs := make(DependencyCureError, len(errs))
for i, v := range errs {
_errs[i] = v.err
}
return _errs
}
// Unwrap returns a deduplicated slice of underlying errors.
func (e *DependencyCureError) Unwrap() []error {
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
defer cancel()
errs := e.unwrap(ctx, NewIR())
_errs := make([]error, len(errs))
for i, err := range errs {
_errs[i] = err
}
return _errs
}
// Error returns a user-facing multiline error message.
func (e *DependencyCureError) Error() string {
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
defer cancel()
ir := NewIR()
errs := e.unwrap(ctx, ir)
if len(errs) == 0 {
return "invalid dependency cure outcome"
}
var buf strings.Builder var buf strings.Builder
buf.WriteString("errors curing dependencies:") buf.WriteString("errors curing inputs:")
for _, err := range errs { for _, i := range p {
buf.WriteString("\n\t" + buf.WriteString("\n\t" +
reportName(err.A, ir.Ident(err.A)) + ": " + reportName(i.a, i.id) + ": " +
err.Error()) e[i.a].Error())
}
if ctx.Err() != nil {
buf.WriteString("\nerror resolution cancelled")
} }
return buf.String() return buf.String()
} }
// Unwrap returns a slice of underlying errors in unspecified order.
func (e InputError) Unwrap() []error {
return slices.AppendSeq(make([]error, 0, len(e)), maps.Values(e))
}
// enterCure must be called before entering an [Artifact] implementation. // enterCure must be called before entering an [Artifact] implementation.
func (c *Cache) enterCure(a Artifact, curesExempt bool) error { func (c *Cache) enterCure(a Artifact, curesExempt bool) error {
if c.attr.Notify != nil { if c.attr.Notify != nil {
@@ -2023,7 +1956,7 @@ func (c *Cache) cureMany(
wg.Add(len(inputs)) wg.Add(len(inputs))
var mask []bool var mask []bool
res := make([]cureRes, len(inputs)) res := make([]cureRes, len(inputs))
errs := make(DependencyCureError, 0, len(inputs)) errs := make(InputError)
var errsMu sync.Mutex var errsMu sync.Mutex
if shallow { if shallow {
mask = make([]bool, len(inputs)) mask = make([]bool, len(inputs))
@@ -2042,13 +1975,13 @@ func (c *Cache) cureMany(
continue continue
} }
} }
pending := pendingArtifactDep{d, &res[i], &errs, &errsMu, &wg} pending := pendingArtifactDep{d, &res[i], errs, &errsMu, &wg}
go pending.cure(c) go pending.cure(c)
} }
wg.Wait() wg.Wait()
if len(errs) > 0 { if len(errs) > 0 {
return mask, &errs return mask, errs
} }
for i, p := range res { for i, p := range res {
if shallow && mask[i] { if shallow && mask[i] {
@@ -2651,7 +2584,11 @@ func (pending *pendingArtifactDep) cure(c *Cache) {
} }
pending.errsMu.Lock() pending.errsMu.Lock()
*pending.errs = append(*pending.errs, &CureError{pending.a, err}) if errs, ok := err.(InputError); ok {
maps.Copy(pending.errs, errs)
} else {
pending.errs[pending.a] = err
}
pending.errsMu.Unlock() pending.errsMu.Unlock()
} }
+44 -64
View File
@@ -11,6 +11,7 @@ import (
"io" "io"
"io/fs" "io/fs"
"log" "log"
"maps"
"net/http" "net/http"
"os" "os"
"path/filepath" "path/filepath"
@@ -861,14 +862,11 @@ func TestCache(t *testing.T) {
cure: func(f *pkg.FContext) error { cure: func(f *pkg.FContext) error {
panic("attempting to cure impossible artifact") panic("attempting to cure impossible artifact")
}, },
}, nil, nil, pkg.WNew, &pkg.DependencyCureError{ }, nil, nil, pkg.WNew, pkg.InputError{
{ failingFile: struct {
A: failingFile, _ []byte
Err: struct { stub.UniqueError
_ []byte }{UniqueError: 0xbad},
stub.UniqueError
}{UniqueError: 0xbad},
},
}}, }},
}) })
@@ -1949,7 +1947,7 @@ errors during scrub:
} }
} }
func TestDependencyCureError(t *testing.T) { func TestInputError(t *testing.T) {
t.Parallel() t.Parallel()
makeIdent := func(ident ...byte) pkg.Artifact { makeIdent := func(ident ...byte) pkg.Artifact {
@@ -1962,56 +1960,24 @@ func TestDependencyCureError(t *testing.T) {
testCases := []struct { testCases := []struct {
name string name string
err pkg.DependencyCureError err pkg.InputError
want string want string
unwrap []error unwrap []error
}{ }{
{"simple", pkg.DependencyCureError{ {"simple", pkg.InputError{
{A: makeIdent(0xff, 9), Err: stub.UniqueError(0xbad09)}, makeIdent(0xff, 9): stub.UniqueError(0xbad09),
{A: makeIdent(0xff, 0), Err: stub.UniqueError(0xbad00)}, makeIdent(0xff, 0): stub.UniqueError(0xbad00),
{A: makeIdent(0xff, 0xf), Err: stub.UniqueError(0xbad0f)}, makeIdent(0xff, 0xf): stub.UniqueError(0xbad0f),
{A: makeIdent(0xff, 1), Err: stub.UniqueError(0xbad01)}, makeIdent(0xff, 1): stub.UniqueError(0xbad01),
}, `errors curing dependencies: }, `errors curing inputs:
_wAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA: unique error 765184 injected by the test suite _wAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA: unique error 765184 injected by the test suite
_wEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA: unique error 765185 injected by the test suite _wEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA: unique error 765185 injected by the test suite
_wkAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA: unique error 765193 injected by the test suite _wkAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA: unique error 765193 injected by the test suite
_w8AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA: unique error 765199 injected by the test suite`, []error{ _w8AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA: unique error 765199 injected by the test suite`, []error{
&pkg.CureError{A: makeIdent(0xff, 0), Err: stub.UniqueError(0xbad00)}, stub.UniqueError(0xbad00),
&pkg.CureError{A: makeIdent(0xff, 1), Err: stub.UniqueError(0xbad01)}, stub.UniqueError(0xbad01),
&pkg.CureError{A: makeIdent(0xff, 9), Err: stub.UniqueError(0xbad09)}, stub.UniqueError(0xbad09),
&pkg.CureError{A: makeIdent(0xff, 0xf), Err: stub.UniqueError(0xbad0f)}, stub.UniqueError(0xbad0f),
}},
{"dedup", pkg.DependencyCureError{
{A: makeIdent(0xff, 9), Err: stub.UniqueError(0xbad09)},
{A: makeIdent(0xff, 0), Err: stub.UniqueError(0xbad00)},
{A: makeIdent(0xff, 0xfd), Err: &pkg.DependencyCureError{
{A: makeIdent(0xff, 9), Err: stub.UniqueError(0xbad09)},
{A: makeIdent(0xff, 0xc), Err: &pkg.DependencyCureError{
{A: makeIdent(0xff, 0xf), Err: stub.UniqueError(0xbad0f)},
{A: makeIdent(0xff, 0), Err: stub.UniqueError(0xbad00)},
}},
{A: makeIdent(0xff, 0), Err: stub.UniqueError(0xbad00)},
{A: makeIdent(0xff, 0), Err: stub.UniqueError(0xbad00)},
}},
{A: makeIdent(0xff, 0xff), Err: &pkg.DependencyCureError{
{A: makeIdent(0xff, 9), Err: stub.UniqueError(0xbad09)},
{A: makeIdent(0xff, 0xc), Err: &pkg.DependencyCureError{
{A: makeIdent(0xff, 0), Err: stub.UniqueError(0xbad00)},
}},
{A: makeIdent(0xff, 0), Err: stub.UniqueError(0xbad00)},
}},
{A: makeIdent(0xff, 0xf), Err: stub.UniqueError(0xbad0f)},
{A: makeIdent(0xff, 1), Err: stub.UniqueError(0xbad01)},
}, `errors curing dependencies:
_wAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA: unique error 765184 injected by the test suite
_wEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA: unique error 765185 injected by the test suite
_wkAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA: unique error 765193 injected by the test suite
_w8AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA: unique error 765199 injected by the test suite`, []error{
&pkg.CureError{A: makeIdent(0xff, 0), Err: stub.UniqueError(0xbad00)},
&pkg.CureError{A: makeIdent(0xff, 1), Err: stub.UniqueError(0xbad01)},
&pkg.CureError{A: makeIdent(0xff, 9), Err: stub.UniqueError(0xbad09)},
&pkg.CureError{A: makeIdent(0xff, 0xf), Err: stub.UniqueError(0xbad0f)},
}}, }},
} }
for _, tc := range testCases { for _, tc := range testCases {
@@ -2022,7 +1988,19 @@ func TestDependencyCureError(t *testing.T) {
t.Errorf("Error:\n%s\nwant\n%s", got, tc.want) t.Errorf("Error:\n%s\nwant\n%s", got, tc.want)
} }
if unwrap := tc.err.Unwrap(); !reflect.DeepEqual(unwrap, tc.unwrap) { unwrap, unwrapM := tc.err.Unwrap(), make(map[error]struct{})
for _, a := range unwrap {
unwrapM[a] = struct{}{}
}
wantUnwrapM := make(map[error]struct{})
for _, a := range tc.unwrap {
wantUnwrapM[a] = struct{}{}
}
if len(unwrap) != len(unwrapM) ||
len(tc.unwrap) != len(wantUnwrapM) ||
!maps.Equal(unwrapM, wantUnwrapM) {
t.Errorf("Unwrap: %#v, want %#v", unwrap, tc.unwrap) t.Errorf("Unwrap: %#v, want %#v", unwrap, tc.unwrap)
} }
}) })
@@ -2055,19 +2033,21 @@ func (a earlyFailureF) Cure(*pkg.FContext) error {
func BenchmarkEarlyDCE(b *testing.B) { func BenchmarkEarlyDCE(b *testing.B) {
msg := message.New(log.New(os.Stderr, "dce: ", 0)) msg := message.New(log.New(os.Stderr, "dce: ", 0))
msg.SwapVerbose(testing.Verbose()) msg.SwapVerbose(testing.Verbose())
c, err := pkg.Open(b.Context(), msg, check.MustAbs(b.TempDir()), nil)
if err != nil {
b.Fatal(err)
}
_, _, err = c.Cure(earlyFailureF(8))
if !errors.Is(err, stub.UniqueError(0xcafe)) {
b.Fatalf("Cure: error = %v", err)
}
c.Close()
dce := err.(*pkg.DependencyCureError)
for b.Loop() { for b.Loop() {
dce.Unwrap() b.StopTimer()
c, err := pkg.Open(b.Context(), msg, check.MustAbs(b.TempDir()), nil)
if err != nil {
b.Fatal(err)
}
b.StartTimer()
_, _, err = c.Cure(earlyFailureF(8))
b.StopTimer()
if !errors.Is(err, stub.UniqueError(0xcafe)) {
b.Fatalf("Cure: error = %v", err)
}
c.Close()
b.StartTimer()
} }
} }