diff --git a/platform/lifecycle/BUILD.bazel b/platform/lifecycle/BUILD.bazel new file mode 100644 index 00000000..2033e6dc --- /dev/null +++ b/platform/lifecycle/BUILD.bazel @@ -0,0 +1,18 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["lifecycle.go"], + importpath = "github.com/uber/submitqueue/platform/lifecycle", + visibility = ["//visibility:public"], +) + +go_test( + name = "go_default_test", + srcs = ["lifecycle_test.go"], + embed = [":go_default_library"], + deps = [ + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/platform/lifecycle/lifecycle.go b/platform/lifecycle/lifecycle.go new file mode 100644 index 00000000..8c76eb24 --- /dev/null +++ b/platform/lifecycle/lifecycle.go @@ -0,0 +1,94 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package lifecycle provides the Component interface and Group type for +// managing ordered start/stop lifecycles. Every runnable subsystem (consumer, +// publisher, server) implements Component; Group composes them into a single +// Component with deterministic ordering and rollback on partial failure. +package lifecycle + +import ( + "context" + "errors" + "fmt" +) + +// Component is anything with a lifecycle. Construct returns one; hosts drive it. +type Component interface { + // Start initializes and starts the component. The context governs the + // start-up phase (e.g. connecting, subscribing); long-running work may + // outlive the context and must be terminated by calling Stop. + Start(ctx context.Context) error + + // Stop gracefully shuts down the component. The context provides a + // deadline for the shutdown; implementations should respect it and + // return promptly when the context is cancelled. + Stop(ctx context.Context) error +} + +// Group runs an ordered list of Components as one Component. +// +// - Start: members in order; if member i fails to start, members i-1…0 are +// stopped in reverse and the error is returned — no half-started state. +// - Stop: members in REVERSE order (work-acceptors drain before the +// connections under them close); errors joined, none swallowed. +type Group struct { + members []Component +} + +// NewGroup creates a Group from the given components. Nil members are silently +// skipped so callers can pass optional components without nil-checking. +func NewGroup(members ...Component) *Group { + filtered := make([]Component, 0, len(members)) + for _, m := range members { + if m != nil { + filtered = append(filtered, m) + } + } + return &Group{members: filtered} +} + +// Start starts all members in order. If any member fails to start, all +// previously started members are stopped in reverse order and the original +// start error is returned. The stop errors from rollback, if any, are joined +// with the start error. +func (g *Group) Start(ctx context.Context) error { + for i, m := range g.members { + if err := m.Start(ctx); err != nil { + // Rollback: stop members i-1…0 in reverse order. + rollbackErr := g.stopRange(ctx, i-1) + return errors.Join(fmt.Errorf("component %d failed to start: %w", i, err), rollbackErr) + } + } + return nil +} + +// Stop stops all members in reverse order. All stop errors are joined so +// none is swallowed; a single member's failure does not prevent the others +// from being stopped. +func (g *Group) Stop(ctx context.Context) error { + return g.stopRange(ctx, len(g.members)-1) +} + +// stopRange stops members from index hi down to 0 (inclusive), collecting +// all errors. A negative hi is a no-op. +func (g *Group) stopRange(ctx context.Context, hi int) error { + var errs []error + for i := hi; i >= 0; i-- { + if err := g.members[i].Stop(ctx); err != nil { + errs = append(errs, fmt.Errorf("component %d failed to stop: %w", i, err)) + } + } + return errors.Join(errs...) +} diff --git a/platform/lifecycle/lifecycle_test.go b/platform/lifecycle/lifecycle_test.go new file mode 100644 index 00000000..d58c29b7 --- /dev/null +++ b/platform/lifecycle/lifecycle_test.go @@ -0,0 +1,182 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package lifecycle + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// spy records the order of Start/Stop calls and can be configured to fail. +type spy struct { + name string + startErr error + stopErr error + log *[]string +} + +func (s *spy) Start(_ context.Context) error { + *s.log = append(*s.log, "start:"+s.name) + return s.startErr +} + +func (s *spy) Stop(_ context.Context) error { + *s.log = append(*s.log, "stop:"+s.name) + return s.stopErr +} + +func TestGroup_StartStop_HappyPath(t *testing.T) { + var log []string + a := &spy{name: "a", log: &log} + b := &spy{name: "b", log: &log} + c := &spy{name: "c", log: &log} + + g := NewGroup(a, b, c) + + require.NoError(t, g.Start(context.Background())) + assert.Equal(t, []string{"start:a", "start:b", "start:c"}, log) + + log = nil + require.NoError(t, g.Stop(context.Background())) + assert.Equal(t, []string{"stop:c", "stop:b", "stop:a"}, log) +} + +func TestGroup_StartRollback_OnFailure(t *testing.T) { + var log []string + a := &spy{name: "a", log: &log} + b := &spy{name: "b", startErr: fmt.Errorf("b broke"), log: &log} + c := &spy{name: "c", log: &log} + + g := NewGroup(a, b, c) + + err := g.Start(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "b broke") + + // a was started and then rolled back; b failed; c was never started + assert.Equal(t, []string{"start:a", "start:b", "stop:a"}, log) +} + +func TestGroup_StartRollback_FirstMemberFails(t *testing.T) { + var log []string + a := &spy{name: "a", startErr: fmt.Errorf("a broke"), log: &log} + b := &spy{name: "b", log: &log} + + g := NewGroup(a, b) + + err := g.Start(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "a broke") + + // Nothing to roll back — a failed on start, b never started + assert.Equal(t, []string{"start:a"}, log) +} + +func TestGroup_StartRollback_JoinsStopErrors(t *testing.T) { + var log []string + a := &spy{name: "a", stopErr: fmt.Errorf("a stop failed"), log: &log} + b := &spy{name: "b", log: &log} + c := &spy{name: "c", startErr: fmt.Errorf("c broke"), log: &log} + + g := NewGroup(a, b, c) + + err := g.Start(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "c broke") + assert.Contains(t, err.Error(), "a stop failed") + + // a and b started, c failed, then b and a rolled back in reverse + assert.Equal(t, []string{"start:a", "start:b", "start:c", "stop:b", "stop:a"}, log) +} + +func TestGroup_Stop_CollectsAllErrors(t *testing.T) { + var log []string + a := &spy{name: "a", stopErr: fmt.Errorf("a stop failed"), log: &log} + b := &spy{name: "b", stopErr: fmt.Errorf("b stop failed"), log: &log} + c := &spy{name: "c", log: &log} + + g := NewGroup(a, b, c) + require.NoError(t, g.Start(context.Background())) + + log = nil + err := g.Stop(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "a stop failed") + assert.Contains(t, err.Error(), "b stop failed") + + // All three stopped in reverse despite errors + assert.Equal(t, []string{"stop:c", "stop:b", "stop:a"}, log) +} + +func TestGroup_NilMembers_Skipped(t *testing.T) { + var log []string + a := &spy{name: "a", log: &log} + + g := NewGroup(nil, a, nil) + + require.NoError(t, g.Start(context.Background())) + assert.Equal(t, []string{"start:a"}, log) + + log = nil + require.NoError(t, g.Stop(context.Background())) + assert.Equal(t, []string{"stop:a"}, log) +} + +func TestGroup_Empty(t *testing.T) { + g := NewGroup() + require.NoError(t, g.Start(context.Background())) + require.NoError(t, g.Stop(context.Background())) +} + +func TestGroup_Nested(t *testing.T) { + var log []string + a := &spy{name: "a", log: &log} + b := &spy{name: "b", log: &log} + c := &spy{name: "c", log: &log} + d := &spy{name: "d", log: &log} + + inner := NewGroup(b, c) + outer := NewGroup(a, inner, d) + + require.NoError(t, outer.Start(context.Background())) + assert.Equal(t, []string{"start:a", "start:b", "start:c", "start:d"}, log) + + log = nil + require.NoError(t, outer.Stop(context.Background())) + assert.Equal(t, []string{"stop:d", "stop:c", "stop:b", "stop:a"}, log) +} + +func TestGroup_Nested_RollbackOnInnerFailure(t *testing.T) { + var log []string + a := &spy{name: "a", log: &log} + b := &spy{name: "b", log: &log} + c := &spy{name: "c", startErr: fmt.Errorf("c broke"), log: &log} + d := &spy{name: "d", log: &log} + + inner := NewGroup(b, c) + outer := NewGroup(a, inner, d) + + err := outer.Start(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "c broke") + + // a started, inner started b then c failed, inner rolled back b, + // then outer rolled back a. d never started. + assert.Equal(t, []string{"start:a", "start:b", "start:c", "stop:b", "stop:a"}, log) +}