From e538ff5c878152e2acfb99f4aed1c77984f41f6b Mon Sep 17 00:00:00 2001 From: Jonathan Ogilvie Date: Fri, 2 Oct 2026 13:50:05 -0400 Subject: [PATCH] fix(render): stop already-started function runtimes when a later one fails to start StartFunctionRuntimes starts each Function's runtime in turn. If getting or starting a later runtime failed it returned the error without stopping the runtimes it had already started, so their containers were leaked and the caller had no FunctionAddresses to stop them with. On a get or start failure, stop the runtimes started so far with StopFunctionRuntimes, which detaches the stop from ctx's cancellation (ctx may be why the start failed). The start error stays primary; any stop errors are joined to it. FunctionAddresses also kept runtimes in a map keyed by Function name, so when two Functions shared a name the first runtime was overwritten and never stopped, on success as well as on rollback. It now keeps every started runtime in a slice; addresses stay keyed by name, last wins. StartFunctionRuntimes is now a thin wrapper over startFunctionRuntimes, which takes the runtime getter as a parameter so the rollback can be tested. Fixes #396 Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Jonathan Ogilvie --- cmd/crossplane/render/render.go | 52 ++++-- cmd/crossplane/render/render_test.go | 245 ++++++++++++++++++++++++--- 2 files changed, 257 insertions(+), 40 deletions(-) diff --git a/cmd/crossplane/render/render.go b/cmd/crossplane/render/render.go index bed3f635..f09ea681 100644 --- a/cmd/crossplane/render/render.go +++ b/cmd/crossplane/render/render.go @@ -95,8 +95,18 @@ type OperationOutputs struct { // FunctionAddresses maps function names to their gRPC target addresses. type FunctionAddresses struct { - addrs map[string]string - contexts map[string]RuntimeContext + addrs map[string]string + // runtimes holds every runtime started, in start order. It's a slice + // rather than a map keyed by name so that Functions sharing a name all + // get stopped, even though only the last one's address is used. + runtimes []startedRuntime +} + +// startedRuntime is a started runtime and the name of its Function. +type startedRuntime struct { + RuntimeContext + + name string } // Addresses returns the function name to gRPC address map. @@ -108,9 +118,9 @@ func (fa *FunctionAddresses) Addresses() map[string]string { // returned error joins all failures. Each runtime bounds its own cleanup. func (fa *FunctionAddresses) Stop(ctx context.Context) error { var errs []error - for name, rctx := range fa.contexts { - if err := rctx.Stop(ctx); err != nil { - errs = append(errs, errors.Wrapf(err, "cannot stop function %q runtime (target %q)", name, rctx.Target)) + for _, rt := range fa.runtimes { + if err := rt.Stop(ctx); err != nil { + errs = append(errs, errors.Wrapf(err, "cannot stop function %q runtime (target %q)", rt.name, rt.Target)) } } return errors.Join(errs...) @@ -120,25 +130,41 @@ func (fa *FunctionAddresses) Stop(ctx context.Context) error { // gRPC addresses. The caller must call Stop on the returned FunctionAddresses // when done. func StartFunctionRuntimes(ctx context.Context, log logging.Logger, fns []pkgv1.Function) (*FunctionAddresses, error) { - addrs := make(map[string]string, len(fns)) - contexts := make(map[string]RuntimeContext, len(fns)) + return startFunctionRuntimes(ctx, log, fns, GetRuntime) +} + +// startFunctionRuntimes implements StartFunctionRuntimes, using getRuntime to +// get each Function's runtime. +func startFunctionRuntimes(ctx context.Context, log logging.Logger, fns []pkgv1.Function, getRuntime func(pkgv1.Function, logging.Logger) (Runtime, error)) (*FunctionAddresses, error) { + fa := &FunctionAddresses{addrs: make(map[string]string, len(fns))} for _, fn := range fns { - rt, err := GetRuntime(fn, log) + rt, err := getRuntime(fn, log) if err != nil { - return nil, errors.Wrapf(err, "cannot get runtime for Function %q", fn.GetName()) + return nil, stopStarted(ctx, fa, errors.Wrapf(err, "cannot get runtime for Function %q", fn.GetName())) } rctx, err := rt.Start(ctx) if err != nil { - return nil, errors.Wrapf(err, "cannot start Function %q", fn.GetName()) + return nil, stopStarted(ctx, fa, errors.Wrapf(err, "cannot start Function %q", fn.GetName())) } - addrs[fn.GetName()] = rctx.Target - contexts[fn.GetName()] = rctx + fa.addrs[fn.GetName()] = rctx.Target + fa.runtimes = append(fa.runtimes, startedRuntime{name: fn.GetName(), RuntimeContext: rctx}) } - return &FunctionAddresses{addrs: addrs, contexts: contexts}, nil + return fa, nil +} + +// stopStarted stops the runtimes started before a later Function failed to +// start. It returns startErr, joined with any stop errors. StopFunctionRuntimes +// detaches the stop from ctx's cancellation, since ctx may be why the start +// failed. +func stopStarted(ctx context.Context, fa *FunctionAddresses, startErr error) error { + if err := StopFunctionRuntimes(ctx, fa); err != nil { + return errors.Join(startErr, err) + } + return startErr } // RewriteAddressesForDocker rewrites function addresses so they are reachable diff --git a/cmd/crossplane/render/render_test.go b/cmd/crossplane/render/render_test.go index c54f44a8..96490396 100644 --- a/cmd/crossplane/render/render_test.go +++ b/cmd/crossplane/render/render_test.go @@ -9,6 +9,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "github.com/crossplane/crossplane-runtime/v2/pkg/errors" + "github.com/crossplane/crossplane-runtime/v2/pkg/logging" pkgv1 "github.com/crossplane/crossplane/apis/v2/pkg/v1" ) @@ -99,12 +100,12 @@ func TestFunctionAddressesStop(t *testing.T) { errA := errors.New("boom a") errB := errors.New("boom b") - stopReturns := func(err error) RuntimeContext { - return RuntimeContext{Target: "fn:9443", Stop: func(context.Context) error { return err }} + stopReturns := func(name string, err error) startedRuntime { + return startedRuntime{name: name, RuntimeContext: RuntimeContext{Target: name + ":9443", Stop: func(context.Context) error { return err }}} } type args struct { - contexts map[string]RuntimeContext + runtimes []startedRuntime } type want struct { // errs are the errors the returned error must wrap. When empty, Stop @@ -120,20 +121,20 @@ func TestFunctionAddressesStop(t *testing.T) { "AllSucceed": { reason: "Stop should return nil when every runtime stops.", args: args{ - contexts: map[string]RuntimeContext{ - "ok-a": stopReturns(nil), - "ok-b": stopReturns(nil), + runtimes: []startedRuntime{ + stopReturns("ok-a", nil), + stopReturns("ok-b", nil), }, }, }, "SomeFail": { reason: "Stop should stop every runtime even if some fail, and return every failure.", args: args{ - contexts: map[string]RuntimeContext{ - "ok-a": stopReturns(nil), - "fail-a": stopReturns(errA), - "ok-b": stopReturns(nil), - "fail-b": stopReturns(errB), + runtimes: []startedRuntime{ + stopReturns("fail-a", errA), + stopReturns("ok-a", nil), + stopReturns("fail-b", errB), + stopReturns("ok-b", nil), }, }, want: want{ @@ -144,21 +145,17 @@ func TestFunctionAddressesStop(t *testing.T) { for name, tc := range cases { t.Run(name, func(t *testing.T) { - // Repeat so Go's randomized map iteration can't hide a regression - // to returning on the first error. - for range 20 { - fa := &FunctionAddresses{contexts: tc.args.contexts} + fa := &FunctionAddresses{runtimes: tc.args.runtimes} - err := fa.Stop(t.Context()) + err := fa.Stop(t.Context()) - wantErrs := tc.want.errs - if len(wantErrs) == 0 { - wantErrs = []error{nil} - } - for _, want := range wantErrs { - if diff := cmp.Diff(want, err, cmpopts.EquateErrors()); diff != "" { - t.Fatalf("\n%s\nStop(...): -want error, +got error:\n%s", tc.reason, diff) - } + wantErrs := tc.want.errs + if len(wantErrs) == 0 { + wantErrs = []error{nil} + } + for _, want := range wantErrs { + if diff := cmp.Diff(want, err, cmpopts.EquateErrors()); diff != "" { + t.Errorf("\n%s\nStop(...): -want error, +got error:\n%s", tc.reason, diff) } } }) @@ -217,11 +214,11 @@ func TestStopFunctionRuntimes(t *testing.T) { var got want var fa *FunctionAddresses if !tc.args.noRuntimes { - fa = &FunctionAddresses{contexts: map[string]RuntimeContext{ - "fn": {Target: "fn:9443", Stop: func(ctx context.Context) error { + fa = &FunctionAddresses{runtimes: []startedRuntime{ + {name: "fn", RuntimeContext: RuntimeContext{Target: "fn:9443", Stop: func(ctx context.Context) error { got.ctxErr = ctx.Err() return tc.args.stopErr - }}, + }}}, }} } @@ -239,3 +236,197 @@ func TestStopFunctionRuntimes(t *testing.T) { }) } } + +type mockRuntime struct { + MockStart func(ctx context.Context) (RuntimeContext, error) +} + +func (m *mockRuntime) Start(ctx context.Context) (RuntimeContext, error) { return m.MockStart(ctx) } + +var _ Runtime = &mockRuntime{} + +func TestStartFunctionRuntimes(t *testing.T) { + errGet := errors.New("get boom") + errStart := errors.New("start boom") + errStopA := errors.New("stop a boom") + errStopB := errors.New("stop b boom") + + // started returns a runtime that starts at the supplied target, and whose + // Stop returns the supplied error. + started := func(target string, stopErr error) Runtime { + return &mockRuntime{MockStart: func(context.Context) (RuntimeContext, error) { + return RuntimeContext{Target: target, Stop: func(context.Context) error { return stopErr }}, nil + }} + } + // startFails returns a runtime that fails to start with the supplied + // error. + startFails := func(err error) Runtime { + return &mockRuntime{MockStart: func(context.Context) (RuntimeContext, error) { + return RuntimeContext{}, err + }} + } + // getRuntimes returns a runtime getter that gets each Function's runtime + // from the supplied map, and returns errGet for a Function not in it. + getRuntimes := func(rts map[string]Runtime) func(pkgv1.Function, logging.Logger) (Runtime, error) { + return func(fn pkgv1.Function, _ logging.Logger) (Runtime, error) { + rt, ok := rts[fn.GetName()] + if !ok { + return nil, errGet + } + return rt, nil + } + } + // getRuntimesInOrder returns a runtime getter that returns the supplied + // runtimes in turn, for Functions that share a name. + getRuntimesInOrder := func(rts ...Runtime) func(pkgv1.Function, logging.Logger) (Runtime, error) { + return func(pkgv1.Function, logging.Logger) (Runtime, error) { + rt := rts[0] + rts = rts[1:] + return rt, nil + } + } + fn := func(name string) pkgv1.Function { + return pkgv1.Function{ObjectMeta: metav1.ObjectMeta{Name: name}} + } + + type args struct { + fns []pkgv1.Function + getRuntime func(pkgv1.Function, logging.Logger) (Runtime, error) + } + type want struct { + // addrs are the addresses of the returned FunctionAddresses, or nil + // if none are returned. + addrs map[string]string + // errs are the errors the returned error must wrap. When empty, + // startFunctionRuntimes must return nil. + errs []error + // stopErrs are the errors stopping the returned FunctionAddresses + // must return, proving which runtimes it stops. When empty, it must + // stop without error. + stopErrs []error + } + + cases := map[string]struct { + reason string + args args + want want + }{ + "StartsAll": { + reason: "startFunctionRuntimes should start every Function's runtime and return their addresses.", + args: args{ + fns: []pkgv1.Function{fn("fn-a"), fn("fn-b")}, + getRuntime: getRuntimes(map[string]Runtime{ + "fn-a": started("fn-a:9443", nil), + "fn-b": started("fn-b:9443", nil), + }), + }, + want: want{ + addrs: map[string]string{"fn-a": "fn-a:9443", "fn-b": "fn-b:9443"}, + }, + }, + "StopsStartedOnStartFailure": { + reason: "startFunctionRuntimes should stop the runtimes it already started when a later one fails to start, and return the start error joined with any stop errors.", + args: args{ + fns: []pkgv1.Function{fn("fn-a"), fn("fn-b"), fn("fn-c")}, + getRuntime: getRuntimes(map[string]Runtime{ + "fn-a": started("fn-a:9443", errStopA), + "fn-b": started("fn-b:9443", errStopB), + "fn-c": startFails(errStart), + }), + }, + want: want{ + errs: []error{errStart, errStopA, errStopB}, + }, + }, + "StopsStartedOnGetRuntimeFailure": { + reason: "startFunctionRuntimes should stop the runtimes it already started when it cannot get a later Function's runtime.", + args: args{ + fns: []pkgv1.Function{fn("fn-a"), fn("fn-c")}, + getRuntime: getRuntimes(map[string]Runtime{ + "fn-a": started("fn-a:9443", errStopA), + }), + }, + want: want{ + errs: []error{errGet, errStopA}, + }, + }, + "StartsDuplicateNames": { + reason: "startFunctionRuntimes should use the last address for Functions that share a name, but keep every runtime so stopping them stops all of them.", + args: args{ + fns: []pkgv1.Function{fn("fn-a"), fn("fn-a")}, + getRuntime: getRuntimesInOrder( + started("fn-a-1:9443", errStopA), + started("fn-a-2:9443", errStopB), + ), + }, + want: want{ + addrs: map[string]string{"fn-a": "fn-a-2:9443"}, + stopErrs: []error{errStopA, errStopB}, + }, + }, + "StopsDuplicateNamesOnStartFailure": { + reason: "startFunctionRuntimes should stop every runtime it already started, including Functions that share a name, when a later one fails to start.", + args: args{ + fns: []pkgv1.Function{fn("fn-a"), fn("fn-a"), fn("fn-c")}, + getRuntime: getRuntimesInOrder( + started("fn-a-1:9443", errStopA), + started("fn-a-2:9443", errStopB), + startFails(errStart), + ), + }, + want: want{ + errs: []error{errStart, errStopA, errStopB}, + }, + }, + "FirstStartFailure": { + reason: "startFunctionRuntimes should return the start error when the first Function fails to start.", + args: args{ + fns: []pkgv1.Function{fn("fn-c")}, + getRuntime: getRuntimes(map[string]Runtime{ + "fn-c": startFails(errStart), + }), + }, + want: want{ + errs: []error{errStart}, + }, + }, + } + + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + fa, err := startFunctionRuntimes(t.Context(), logging.NewNopLogger(), tc.args.fns, tc.args.getRuntime) + + var addrs map[string]string + if fa != nil { + addrs = fa.Addresses() + } + if diff := cmp.Diff(tc.want.addrs, addrs); diff != "" { + t.Errorf("\n%s\nstartFunctionRuntimes(...): -want addresses, +got addresses:\n%s", tc.reason, diff) + } + + wantErrs := tc.want.errs + if len(wantErrs) == 0 { + wantErrs = []error{nil} + } + for _, want := range wantErrs { + if diff := cmp.Diff(want, err, cmpopts.EquateErrors()); diff != "" { + t.Errorf("\n%s\nstartFunctionRuntimes(...): -want error, +got error:\n%s", tc.reason, diff) + } + } + + if fa == nil { + return + } + stopErr := fa.Stop(t.Context()) + wantStopErrs := tc.want.stopErrs + if len(wantStopErrs) == 0 { + wantStopErrs = []error{nil} + } + for _, want := range wantStopErrs { + if diff := cmp.Diff(want, stopErr, cmpopts.EquateErrors()); diff != "" { + t.Errorf("\n%s\nfa.Stop(...): -want error, +got error:\n%s", tc.reason, diff) + } + } + }) + } +}