From 5cc9c6ee3f8591e5bdce5334f7ebbfe1922f6fa2 Mon Sep 17 00:00:00 2001 From: SirLouen Date: Tue, 3 Nov 2026 22:22:46 +0100 Subject: [PATCH 1/3] docs(pluginkit): say what a host may ask of a plugin's Start, Stop, Migrate and Routes --- pluginkit/pluginkit.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/pluginkit/pluginkit.go b/pluginkit/pluginkit.go index 40f27ab..16f3ec9 100644 --- a/pluginkit/pluginkit.go +++ b/pluginkit/pluginkit.go @@ -13,14 +13,16 @@ import ( // managed lifecycle. type Plugin interface { ID() string + // Start readies the plugin, returning by the time ctx ends, and may run again after Stop. Start(ctx context.Context) error - // Stop releases what the plugin holds, returning by the time ctx ends, whether or not Start ran. + // Stop releases what the plugin holds by the time ctx ends, whether or not Start ran, and may run more than once. Stop(ctx context.Context) error } // Migrator is implemented by plugins that own database schema, which a host migrates before starting any // plugin, and which a host with a [MigrationStore] skips while its recorded version is current. type Migrator interface { + // Migrate applies the plugin's schema, returning by the time ctx ends. Migrate(ctx context.Context) error } @@ -50,11 +52,13 @@ type Seeder interface { // RouteProvider is implemented by plugins that expose HTTP endpoints // under their own namespace. type RouteProvider interface { + // Routes returns the plugin's handler, whether or not Start ran or worked. Routes() http.Handler } // PublicPathProvider is implemented by plugins declaring namespace-relative paths that must stay // reachable without a session. Paths match exactly, for every HTTP method, and all else stays protected. type PublicPathProvider interface { + // PublicPaths returns those paths, whether or not Start ran or worked. PublicPaths() []string } From 22a95430f43b8488ca2d1d35abee2e1541a61aa6 Mon Sep 17 00:00:00 2001 From: SirLouen Date: Tue, 3 Nov 2026 22:22:46 +0100 Subject: [PATCH 2/3] feat(pluginkit): start each plugin on its own and report the ones that failed --- pluginkit/CHANGELOG.md | 4 +- pluginkit/host.go | 64 ++-- pluginkit/seed_test.go | 30 ++ pluginkit/start_each.go | 93 +++++ pluginkit/start_each_context_test.go | 315 +++++++++++++++++ pluginkit/start_each_test.go | 492 +++++++++++++++++++++++++++ pluginkit/store_test.go | 14 + 7 files changed, 979 insertions(+), 33 deletions(-) create mode 100644 pluginkit/start_each.go create mode 100644 pluginkit/start_each_context_test.go create mode 100644 pluginkit/start_each_test.go diff --git a/pluginkit/CHANGELOG.md b/pluginkit/CHANGELOG.md index 6304af7..91bec6a 100644 --- a/pluginkit/CHANGELOG.md +++ b/pluginkit/CHANGELOG.md @@ -6,8 +6,10 @@ Releases of this module are tagged `pluginkit/vX.Y.Z`. Releases up to ## Unreleased - The module needs Go 1.27.3. -- `NewHostWithMigrationStore` returns a host whose `Migrate` and `Start` skip each plugin a `MigrationStore` records as current, by the version its `MigrationVersioner` reports. +- `NewHostWithMigrationStore` returns a host whose `Migrate`, `Start` and `StartEach` skip each plugin a `MigrationStore` records as current, by the version its `MigrationVersioner` reports. - An optional `MigratorOf` lookup gives that host the `Migrator` of a plugin that ships its schema as data. +- `Host.StartEach` migrates and then starts every plugin on its own, goes on past a failure or a panic, and answers a `StartOutcome` naming each plugin that started and each `Failure`. +- The `Plugin` doc promises that `Stop` may run more than once and that `Start` may run again after it, and asks `Start` to return by the time its context ends. ## 0.7.0 - 2026-10-15 diff --git a/pluginkit/host.go b/pluginkit/host.go index c42d96c..954c6ba 100644 --- a/pluginkit/host.go +++ b/pluginkit/host.go @@ -15,6 +15,8 @@ var ( errNoMigrationStore = errors.New("pluginkit: the migration store is nil") // errEmptyMigrationVersion refuses a migration version that holds nothing to compare. errEmptyMigrationVersion = errors.New("the version is empty") + // errNilMigrator refuses a lookup that claims a migrator and gives none. + errNilMigrator = errors.New("the lookup gave a nil migrator") ) // Host starts and stops a fixed set of plugins. @@ -33,7 +35,7 @@ func NewHost(plugins ...Plugin) *Host { return &Host{plugins: plugins} } -// NewHostWithMigrationStore returns a [Host] whose Migrate skips every plugin store records as current. +// NewHostWithMigrationStore returns a [Host] whose migrations skip every plugin store records as current. func NewHostWithMigrationStore(store MigrationStore, migratorOf MigratorOf, plugins ...Plugin) (*Host, error) { if store == nil { return nil, errNoMigrationStore @@ -79,26 +81,26 @@ func (h *Host) rollBack(ctx context.Context, stopGrace time.Duration, index int) return h.stopDownFrom(stopping, index) } -// Migrate applies the schema of every [Migrator] plugin in order, past those its store records as current. +// Migrate applies every [Migrator] plugin's schema in order, past those its store records as current, until one fails. func (h *Host) Migrate(ctx context.Context) error { - if h.store == nil { - return h.migrateEvery(ctx) - } versions, err := h.readVersions(ctx) if err != nil { return err } for _, p := range h.plugins { - if err := h.migrateUnlessCurrent(ctx, p, versions); err != nil { + if _, err := h.migrateUnlessCurrent(ctx, p, versions); err != nil { return err } } return nil } -// readVersions returns every migration version the store records, in one read. +// readVersions returns every migration version the store records, in one read, and none on a host with no store. func (h *Host) readVersions(ctx context.Context) (map[string]string, error) { var versions map[string]string + if h.store == nil { + return versions, nil + } err := guarded(ctx, "read the migration versions", func(ctx context.Context) error { var err error versions, err = h.store.Versions(ctx) @@ -107,24 +109,24 @@ func (h *Host) readVersions(ctx context.Context) (map[string]string, error) { return versions, err } -// migrateUnlessCurrent applies the schema of p unless versions holds its version, then records the one it applied. -func (h *Host) migrateUnlessCurrent(ctx context.Context, p Plugin, versions map[string]string) error { +// migrateUnlessCurrent applies and records p's schema unless versions holds it, answering the step an error came from. +func (h *Host) migrateUnlessCurrent(ctx context.Context, p Plugin, versions map[string]string) (Step, error) { migrator, ok, err := h.migratorFor(ctx, p) if err != nil || !ok { - return err + return StepMigrate, err } versioner, ok := migrator.(MigrationVersioner) - if !ok { - return safeCall(ctx, p.ID(), "migrate", migrator.Migrate) + if !ok || h.store == nil { + return StepMigrate, safeCall(ctx, p.ID(), "migrate", migrator.Migrate) } version, err := migrationVersion(ctx, p.ID(), versioner) if err != nil || version == versions[p.ID()] { - return err + return StepMigrate, err } if err := safeCall(ctx, p.ID(), "migrate", migrator.Migrate); err != nil { - return err + return StepMigrate, err } - return safeCall(ctx, p.ID(), "record", func(ctx context.Context) error { + return StepRecord, safeCall(ctx, p.ID(), "record", func(ctx context.Context) error { return h.store.Record(ctx, p.ID(), version) }) } @@ -134,6 +136,9 @@ func (h *Host) migratorFor(ctx context.Context, p Plugin) (migrator Migrator, ok if h.migratorOf != nil { err = safeCall(ctx, p.ID(), "migrator lookup", func(context.Context) error { migrator, ok = h.migratorOf(p) + if ok && migrator == nil { + return errNilMigrator + } return nil }) if err != nil || ok { @@ -157,28 +162,15 @@ func migrationVersion(ctx context.Context, id string, versioner MigrationVersion return version, err } -// migrateEvery applies the schema of every [Migrator] plugin in registration order, stopping at the first failure. -func (h *Host) migrateEvery(ctx context.Context) error { - return callEach(ctx, h.plugins, "migrate", Migrator.Migrate) -} - // Seed asks every [Seeder] plugin to fill its schema in registration order, // stopping at the first failure. func (h *Host) Seed(ctx context.Context) error { - return callEach(ctx, h.plugins, "seed", Seeder.Seed) -} - -// callEach calls step on every plugin that is a T, in registration order, stopping at the first failure. -func callEach[T any]( - ctx context.Context, plugins []Plugin, operation string, step func(T, context.Context) error, -) error { - for _, p := range plugins { - capable, ok := p.(T) + for _, p := range h.plugins { + seeder, ok := p.(Seeder) if !ok { continue } - call := func(ctx context.Context) error { return step(capable, ctx) } - if err := safeCall(ctx, p.ID(), operation, call); err != nil { + if err := safeCall(ctx, p.ID(), "seed", seeder.Seed); err != nil { return err } } @@ -236,7 +228,7 @@ func safeCall(ctx context.Context, id, operation string, fn func(context.Context func guarded(ctx context.Context, name string, fn func(context.Context) error) (err error) { defer func() { if recovered := recover(); recovered != nil { - err = fmt.Errorf("pluginkit: %s panicked: %v", name, recovered) + err = panicError(fmt.Sprintf("pluginkit: %s panicked: %v", name, recovered)) } }() if err := fn(ctx); err != nil { @@ -244,3 +236,11 @@ func guarded(ctx context.Context, name string, fn func(context.Context) error) ( } return nil } + +// panicError is a panic [guarded] recovered, holding the text that names where it came from. +type panicError string + +// Error returns the text of the panic. +func (e panicError) Error() string { + return string(e) +} diff --git a/pluginkit/seed_test.go b/pluginkit/seed_test.go index 7ffabd9..a6f4e8b 100644 --- a/pluginkit/seed_test.go +++ b/pluginkit/seed_test.go @@ -85,6 +85,36 @@ func TestHostTagsASeedPanicWithItsPlugin(t *testing.T) { } } +// contextSeeder is a seeder that records the caller value its context carries and whether that context can end. +type contextSeeder struct { + fakePlugin + seen *[]any +} + +// Seed records the caller value of ctx and whether ctx can end. +func (c *contextSeeder) Seed(ctx context.Context) error { + *c.seen = append(*c.seen, ctx.Value(callerKey{}), ctx.Done() != nil) + return nil +} + +func TestHostSeedHandsTheCallerContextToEverySeeder(t *testing.T) { + t.Parallel() + + var calls []string + var seen []any + host := pluginkit.NewHost(&contextSeeder{fakePlugin: fakePlugin{id: "alpha", calls: &calls}, seen: &seen}) + ctx, cancel := context.WithCancel(context.WithValue(t.Context(), callerKey{}, "caller")) + defer cancel() + + if err := host.Seed(ctx); err != nil { + t.Fatalf("Seed() error = %v, want nil", err) + } + + if want := []any{"caller", true}; !slices.Equal(want, seen) { + t.Errorf("context seen = %v, want %v", seen, want) + } +} + func TestHostSeedsNothingWithoutASeeder(t *testing.T) { t.Parallel() diff --git a/pluginkit/start_each.go b/pluginkit/start_each.go new file mode 100644 index 0000000..1bb2fc9 --- /dev/null +++ b/pluginkit/start_each.go @@ -0,0 +1,93 @@ +// SPDX-License-Identifier: Apache-2.0 + +package pluginkit + +import ( + "context" + "time" +) + +// Step names the part of a plugin's boot a [Failure] happened in. +type Step string + +const ( + // StepMigrate is the migration pass, and a plugin that fails it does not start. + StepMigrate Step = "migrate" + // StepRecord is the note of a migration that ran, and a plugin that fails it still starts. + StepRecord Step = "record" + // StepStart is a plugin's Start. + StepStart Step = "start" +) + +// Failure is one step one plugin failed, or was still waiting for when the context ended. +type Failure struct { + ID string + Step Step + // Panicked is true only for a panic this step recovered, never for an error that merely holds one. + Panicked bool + Err error +} + +// StartedPlugin is one plugin whose Start returned nil, and how long that Start took. +type StartedPlugin struct { + ID string + Took time.Duration +} + +// StartOutcome is what [Host.StartEach] answers, both lists in registration order, a failed record beside its start. +type StartOutcome struct { + Started []StartedPlugin + Failed []Failure +} + +// StartEach migrates then starts each plugin on its own past any failure, stopping none, and errs only on a store read. +func (h *Host) StartEach(ctx context.Context) (StartOutcome, error) { + versions, err := h.readVersions(ctx) + if err != nil { + return StartOutcome{}, err + } + migrations := make([]Failure, len(h.plugins)) + for i, p := range h.plugins { + migrations[i] = h.migrateUnlessEnded(ctx, p, versions) + } + var outcome StartOutcome + for i, p := range h.plugins { + outcome.startAfter(ctx, p, migrations[i]) + } + return outcome, nil +} + +// migrateUnlessEnded migrates p while ctx lasts, answering a [Failure] whose Err is nil when nothing failed. +func (h *Host) migrateUnlessEnded(ctx context.Context, p Plugin, versions map[string]string) Failure { + if err := safeCall(ctx, p.ID(), string(StepMigrate), context.Context.Err); err != nil { + return failureOf(p.ID(), StepMigrate, err) + } + step, err := h.migrateUnlessCurrent(ctx, p, versions) + return failureOf(p.ID(), step, err) +} + +// startAfter lists what p's migration did when it failed, then starts p unless that failure rules it out. +func (o *StartOutcome) startAfter(ctx context.Context, p Plugin, migration Failure) { + if migration.Err != nil { + o.Failed = append(o.Failed, migration) + } + if migration.Err != nil && migration.Step != StepRecord { + return + } + began := time.Now() + err := safeCall(ctx, p.ID(), string(StepStart), context.Context.Err) + if err == nil { + err = safeCall(ctx, p.ID(), string(StepStart), p.Start) + } + if err != nil { + o.Failed = append(o.Failed, failureOf(p.ID(), StepStart, err)) + return + } + o.Started = append(o.Started, StartedPlugin{ID: p.ID(), Took: time.Since(began)}) +} + +// failureOf returns the failure of the plugin id at step, marked as panicked only when err is that step's own panic. +func failureOf(id string, step Step, err error) Failure { + _, panicked := err.(panicError) + return Failure{ID: id, Step: step, Panicked: panicked, Err: err} +} diff --git a/pluginkit/start_each_context_test.go b/pluginkit/start_each_context_test.go new file mode 100644 index 0000000..678a57e --- /dev/null +++ b/pluginkit/start_each_context_test.go @@ -0,0 +1,315 @@ +// SPDX-License-Identifier: Apache-2.0 + +package pluginkit_test + +import ( + "context" + "errors" + "slices" + "testing" + "time" + + "github.com/gopherium/framework/pluginkit" +) + +// cancellingMigrator is a plugin whose Migrate ends the startup context and succeeds. +type cancellingMigrator struct { + fakePlugin + cancel context.CancelFunc +} + +// Migrate ends the startup context. +func (c *cancellingMigrator) Migrate(context.Context) error { + *c.calls = append(*c.calls, c.id+" migrate") + c.cancel() + return nil +} + +// cancellingVersionedMigrator is a cancelling migrator whose schema reports a version. +type cancellingVersionedMigrator struct { + cancellingMigrator +} + +// MigrationVersion returns a version no store has recorded. +func (c *cancellingVersionedMigrator) MigrationVersion() (string, error) { + return "1", nil +} + +// succeedingCanceller is a plugin whose Start ends the startup context and succeeds. +type succeedingCanceller struct { + fakePlugin + cancel context.CancelFunc +} + +// Start ends the startup context. +func (s *succeedingCanceller) Start(context.Context) error { + *s.calls = append(*s.calls, s.id+" start") + s.cancel() + return nil +} + +// contextEnded returns the failure of the plugin id that waited for step when the context was cancelled. +func contextEnded(id string, step pluginkit.Step) failureWant { + return failureWant{id, step, false, "pluginkit: " + id + " " + string(step) + ": context canceled"} +} + +func TestStartEachStillRecordsAMigrationThatEndedTheContext(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + recordErr error + failed []failureWant + }{ + {"a record that works", nil, []failureWant{contextEnded("alpha", pluginkit.StepStart)}}, + {"a record that fails", errStoreOffline, []failureWant{ + {"alpha", pluginkit.StepRecord, false, "pluginkit: alpha record: store offline"}, + contextEnded("alpha", pluginkit.StepStart), + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + var calls []string + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + store := &memoryStore{versions: map[string]string{}, recordErr: tc.recordErr, calls: &calls} + alpha := &cancellingVersionedMigrator{ + cancellingMigrator{fakePlugin: fakePlugin{id: "alpha", calls: &calls}, cancel: cancel}, + } + + outcome := startEach(ctx, t, storeHost(t, store, nil, alpha)) + + wantOutcome(t, outcome, nil, tc.failed) + wantCalls(t, calls, []string{"versions", "alpha migrate", "record alpha 1"}) + }) + } +} + +func TestStartEachReportsTheRestWhenTheContextEnds(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + plugins func(c *[]string, cancel context.CancelFunc) []pluginkit.Plugin + calls []string + started []string + failed []failureWant + }{ + {"during a start", func(c *[]string, cancel context.CancelFunc) []pluginkit.Plugin { + return []pluginkit.Plugin{ + plain("alpha", c), + &cancellingStarter{fakePlugin: fakePlugin{id: "beta", calls: c}, cancel: cancel}, + migrating("gamma", c), + } + }, []string{"gamma migrate", "alpha start", "beta start"}, []string{"alpha"}, []failureWant{ + contextEnded("beta", pluginkit.StepStart), contextEnded("gamma", pluginkit.StepStart), + }}, + {"during a start that still succeeds", func(c *[]string, cancel context.CancelFunc) []pluginkit.Plugin { + return []pluginkit.Plugin{ + &succeedingCanceller{fakePlugin: fakePlugin{id: "alpha", calls: c}, cancel: cancel}, + plain("beta", c), + } + }, []string{"alpha start"}, []string{"alpha"}, []failureWant{contextEnded("beta", pluginkit.StepStart)}}, + {"during a migration", func(c *[]string, cancel context.CancelFunc) []pluginkit.Plugin { + return []pluginkit.Plugin{ + &cancellingMigrator{fakePlugin: fakePlugin{id: "alpha", calls: c}, cancel: cancel}, + migrating("beta", c), + plain("gamma", c), + } + }, []string{"alpha migrate"}, nil, []failureWant{ + contextEnded("alpha", pluginkit.StepStart), + contextEnded("beta", pluginkit.StepMigrate), + contextEnded("gamma", pluginkit.StepMigrate), + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + var calls []string + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + host := pluginkit.NewHost(tc.plugins(&calls, cancel)...) + + outcome := startEach(ctx, t, host) + + wantOutcome(t, outcome, tc.started, tc.failed) + wantCalls(t, calls, tc.calls) + for _, failure := range outcome.Failed { + if !errors.Is(failure.Err, context.Canceled) { + t.Errorf("failure of %s = %v, want the context's error in its chain", failure.ID, failure.Err) + } + } + }) + } +} + +func TestStartEachCallsNoPluginOnAContextAlreadyEnded(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithDeadline(t.Context(), time.Now().Add(-time.Minute)) + defer cancel() + var calls []string + store := &memoryStore{versions: map[string]string{}, calls: &calls} + lookup := func(pluginkit.Plugin) (pluginkit.Migrator, bool) { + calls = append(calls, "lookup") + return nil, false + } + host := storeHost(t, store, lookup, versioned("alpha", "1", &calls), plain("beta", &calls)) + + outcome := startEach(ctx, t, host) + + wantOutcome(t, outcome, nil, []failureWant{ + {"alpha", pluginkit.StepMigrate, false, "pluginkit: alpha migrate: context deadline exceeded"}, + {"beta", pluginkit.StepMigrate, false, "pluginkit: beta migrate: context deadline exceeded"}, + }) + wantCalls(t, calls, []string{"versions"}) + for _, failure := range outcome.Failed { + if !errors.Is(failure.Err, context.DeadlineExceeded) { + t.Errorf("failure of %s = %v, want the context's error in its chain", failure.ID, failure.Err) + } + } +} + +func TestStartEachReportsTheContextsErrorOverItsCause(t *testing.T) { + t.Parallel() + + errShutdown := errors.New("the operator asked to stop") + ctx, cancel := context.WithCancelCause(t.Context()) + cancel(errShutdown) + var calls []string + host := pluginkit.NewHost(plain("alpha", &calls)) + + outcome := startEach(ctx, t, host) + + wantOutcome(t, outcome, nil, []failureWant{contextEnded("alpha", pluginkit.StepMigrate)}) + if !errors.Is(outcome.Failed[0].Err, context.Canceled) { + t.Errorf("failure = %v, want the context's error in its chain", outcome.Failed[0].Err) + } +} + +func TestHostStartLeavesAnEndedContextToThePlugins(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + host func(t *testing.T, calls *[]string) *pluginkit.Host + calls []string + }{ + {"with no store", func(_ *testing.T, c *[]string) *pluginkit.Host { + return pluginkit.NewHost(migrating("alpha", c), plain("beta", c)) + }, []string{"alpha migrate", "alpha start", "beta start"}}, + {"with a store", func(t *testing.T, c *[]string) *pluginkit.Host { + store := &memoryStore{versions: map[string]string{}, calls: c} + return storeHost(t, store, nil, versioned("alpha", "1", c), plain("beta", c)) + }, []string{"versions", "alpha version", "alpha migrate", "record alpha 1", "alpha start", "beta start"}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + var calls []string + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + if err := tc.host(t, &calls).Start(ctx, stopGrace); err != nil { + t.Fatalf("Start() error = %v, want nil from plugins that ignore their context", err) + } + + wantCalls(t, calls, tc.calls) + }) + } +} + +// contextStarter is a plugin that records the caller value its Start context carries. +type contextStarter struct { + fakePlugin + seen *[]any +} + +// Start records the caller value of ctx. +func (c *contextStarter) Start(ctx context.Context) error { + *c.seen = append(*c.seen, ctx.Value(callerKey{})) + return nil +} + +func TestStartEachHandsTheCallerContextToEveryStep(t *testing.T) { + t.Parallel() + + var calls []string + var seen []any + host, err := pluginkit.NewHostWithMigrationStore(&contextStore{seen: &seen}, nil, + &versionedContextMigrator{contextMigrator{fakePlugin: fakePlugin{id: "alpha", calls: &calls}, seen: &seen}}, + &contextStarter{fakePlugin: fakePlugin{id: "beta", calls: &calls}, seen: &seen}) + if err != nil { + t.Fatalf("NewHostWithMigrationStore() error = %v, want nil", err) + } + + startEach(context.WithValue(t.Context(), callerKey{}, "caller"), t, host) + + if want := []any{"caller", "caller", "caller", "caller"}; !slices.Equal(want, seen) { + t.Errorf("context values seen = %v, want %v from the read, the migration, the record and the start", seen, want) + } +} + +// endableStore is a migration store that records, for each context it gets, whether that context can end. +type endableStore struct { + seen *[]bool +} + +// Versions records whether ctx can end and returns no versions. +func (s *endableStore) Versions(ctx context.Context) (map[string]string, error) { + *s.seen = append(*s.seen, ctx.Done() != nil) + return map[string]string{}, nil +} + +// Record records whether ctx can end. +func (s *endableStore) Record(ctx context.Context, _, _ string) error { + *s.seen = append(*s.seen, ctx.Done() != nil) + return nil +} + +// endablePlugin is a versioned migrator that records, for each context it gets, whether that context can end. +type endablePlugin struct { + fakePlugin + seen *[]bool +} + +// Migrate records whether ctx can end. +func (e *endablePlugin) Migrate(ctx context.Context) error { + *e.seen = append(*e.seen, ctx.Done() != nil) + return nil +} + +// MigrationVersion returns a version no store has recorded. +func (e *endablePlugin) MigrationVersion() (string, error) { + return "1", nil +} + +// Start records whether ctx can end. +func (e *endablePlugin) Start(ctx context.Context) error { + *e.seen = append(*e.seen, ctx.Done() != nil) + return nil +} + +func TestStartEachHandsEveryStepAContextThatCanEnd(t *testing.T) { + t.Parallel() + + var calls []string + var seen []bool + host, err := pluginkit.NewHostWithMigrationStore(&endableStore{seen: &seen}, nil, + &endablePlugin{fakePlugin: fakePlugin{id: "alpha", calls: &calls}, seen: &seen}) + if err != nil { + t.Fatalf("NewHostWithMigrationStore() error = %v, want nil", err) + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + startEach(ctx, t, host) + + if want := []bool{true, true, true, true}; !slices.Equal(want, seen) { + t.Errorf("can end = %v, want %v from the read, the migration, the record and the start", seen, want) + } +} diff --git a/pluginkit/start_each_test.go b/pluginkit/start_each_test.go new file mode 100644 index 0000000..9a374fd --- /dev/null +++ b/pluginkit/start_each_test.go @@ -0,0 +1,492 @@ +// SPDX-License-Identifier: Apache-2.0 + +package pluginkit_test + +import ( + "context" + "errors" + "slices" + "testing" + "testing/synctest" + "time" + + "github.com/gopherium/framework/pluginkit" +) + +// failureWant is what a test expects of one failure in a start outcome. +type failureWant struct { + id string + step pluginkit.Step + panicked bool + text string +} + +// plain returns a plugin with id that owns no schema. +func plain(id string, calls *[]string) *fakePlugin { + return &fakePlugin{id: id, calls: calls} +} + +// startEach runs StartEach on host and fails t when it answers an error of its own. +func startEach(ctx context.Context, t *testing.T, host *pluginkit.Host) pluginkit.StartOutcome { + t.Helper() + outcome, err := host.StartEach(ctx) + if err != nil { + t.Fatalf("StartEach() error = %v, want nil", err) + } + return outcome +} + +// wantOutcome fails t unless outcome lists the plugins started and the failures failed, in that order. +func wantOutcome(t *testing.T, outcome pluginkit.StartOutcome, started []string, failed []failureWant) { + t.Helper() + ids := make([]string, 0, len(outcome.Started)) + for _, plugin := range outcome.Started { + ids = append(ids, plugin.ID) + } + if !slices.Equal(started, ids) { + t.Errorf("started = %v, want %v", ids, started) + } + got := make([]failureWant, 0, len(outcome.Failed)) + for _, failure := range outcome.Failed { + got = append(got, failureWant{failure.ID, failure.Step, failure.Panicked, failure.Err.Error()}) + } + if !slices.Equal(failed, got) { + t.Errorf("failed = %+v, want %+v", got, failed) + } +} + +func TestStartEachStartsEveryPluginInRegistrationOrder(t *testing.T) { + t.Parallel() + + var calls []string + host := pluginkit.NewHost(plain("alpha", &calls), plain("beta", &calls), plain("gamma", &calls)) + + outcome := startEach(t.Context(), t, host) + + wantOutcome(t, outcome, []string{"alpha", "beta", "gamma"}, nil) + wantCalls(t, calls, []string{"alpha start", "beta start", "gamma start"}) +} + +func TestStartEachMigratesEveryPluginBeforeStartingAny(t *testing.T) { + t.Parallel() + + var calls []string + host := pluginkit.NewHost(plain("alpha", &calls), migrating("beta", &calls), migrating("gamma", &calls)) + + outcome := startEach(t.Context(), t, host) + + wantOutcome(t, outcome, []string{"alpha", "beta", "gamma"}, nil) + wantCalls(t, calls, []string{"beta migrate", "gamma migrate", "alpha start", "beta start", "gamma start"}) +} + +func TestStartEachWithNoPluginsReportsNothing(t *testing.T) { + t.Parallel() + + outcome := startEach(t.Context(), t, pluginkit.NewHost()) + + wantOutcome(t, outcome, nil, nil) +} + +func TestStartEachGoesOnPastEachFailure(t *testing.T) { + t.Parallel() + + errBoot := errors.New("boot failed") + errSchema := errors.New("schema exploded") + var discarded []string + errNested := pluginkit.NewHost(&fakePlugin{id: "inner", startPanic: true, calls: &discarded}). + Start(t.Context(), stopGrace) + failingMigration := func(id string, c *[]string) *migratingPlugin { + plugin := migrating(id, c) + plugin.migrateErr = errSchema + return plugin + } + cases := []struct { + name string + plugins func(c *[]string) []pluginkit.Plugin + calls []string + started []string + failed []failureWant + }{ + {"a failed start", func(c *[]string) []pluginkit.Plugin { + return []pluginkit.Plugin{ + plain("alpha", c), &fakePlugin{id: "beta", startErr: errBoot, calls: c}, plain("gamma", c), + } + }, []string{"alpha start", "beta start", "gamma start"}, []string{"alpha", "gamma"}, []failureWant{ + {"beta", pluginkit.StepStart, false, "pluginkit: beta start: boot failed"}, + }}, + {"a panicking start", func(c *[]string) []pluginkit.Plugin { + return []pluginkit.Plugin{ + plain("alpha", c), &fakePlugin{id: "beta", startPanic: true, calls: c}, plain("gamma", c), + } + }, []string{"alpha start", "beta start", "gamma start"}, []string{"alpha", "gamma"}, []failureWant{ + {"beta", pluginkit.StepStart, true, "pluginkit: beta start panicked: boom"}, + }}, + {"a failed migration", func(c *[]string) []pluginkit.Plugin { + return []pluginkit.Plugin{failingMigration("alpha", c), migrating("beta", c), plain("gamma", c)} + }, []string{"alpha migrate", "beta migrate", "beta start", "gamma start"}, + []string{"beta", "gamma"}, []failureWant{ + {"alpha", pluginkit.StepMigrate, false, "pluginkit: alpha migrate: schema exploded"}, + }}, + {"a panicking migration", func(c *[]string) []pluginkit.Plugin { + panicking := migrating("alpha", c) + panicking.migratePanic = true + return []pluginkit.Plugin{panicking, migrating("beta", c)} + }, []string{"alpha migrate", "beta migrate", "beta start"}, []string{"beta"}, []failureWant{ + {"alpha", pluginkit.StepMigrate, true, "pluginkit: alpha migrate panicked: boom"}, + }}, + {"failures of both steps, in registration order", func(c *[]string) []pluginkit.Plugin { + return []pluginkit.Plugin{ + &fakePlugin{id: "alpha", startErr: errBoot, calls: c}, + failingMigration("beta", c), + &fakePlugin{id: "gamma", startErr: errBoot, calls: c}, + } + }, []string{"beta migrate", "alpha start", "gamma start"}, nil, []failureWant{ + {"alpha", pluginkit.StepStart, false, "pluginkit: alpha start: boot failed"}, + {"beta", pluginkit.StepMigrate, false, "pluginkit: beta migrate: schema exploded"}, + {"gamma", pluginkit.StepStart, false, "pluginkit: gamma start: boot failed"}, + }}, + {"an error holding a panic recovered further in", func(c *[]string) []pluginkit.Plugin { + return []pluginkit.Plugin{&fakePlugin{id: "alpha", startErr: errNested, calls: c}} + }, []string{"alpha start"}, nil, []failureWant{ + {"alpha", pluginkit.StepStart, false, "pluginkit: alpha start: pluginkit: inner start panicked: boom"}, + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + var calls []string + host := pluginkit.NewHost(tc.plugins(&calls)...) + + outcome := startEach(t.Context(), t, host) + + wantOutcome(t, outcome, tc.started, tc.failed) + wantCalls(t, calls, tc.calls) + }) + } +} + +func TestStartEachKeepsEachCauseInItsFailure(t *testing.T) { + t.Parallel() + + errBoot := errors.New("boot failed") + errSchema := errors.New("schema exploded") + var calls []string + store := &memoryStore{versions: map[string]string{}, recordErr: errStoreOffline, calls: &calls} + schema := versioned("alpha", "1", &calls) + schema.migrateErr = errSchema + host := storeHost(t, store, nil, schema, versioned("beta", "1", &calls), + &fakePlugin{id: "gamma", startErr: errBoot, calls: &calls}) + + outcome := startEach(t.Context(), t, host) + + want := []error{errSchema, errStoreOffline, errBoot} + if len(outcome.Failed) != len(want) { + t.Fatalf("failed = %+v, want %d failures", outcome.Failed, len(want)) + } + for i, cause := range want { + if !errors.Is(outcome.Failed[i].Err, cause) { + t.Errorf("failure %d error = %v, want %v in its chain", i, outcome.Failed[i].Err, cause) + } + } +} + +func TestStartEachWithAMigrationStore(t *testing.T) { + t.Parallel() + + errUnreadable := errors.New("folder unreadable") + errBoot := errors.New("boot failed") + errSchema := errors.New("schema exploded") + cases := []struct { + name string + prepare func(store *memoryStore, alpha *versionedPlugin) + calls []string + started []string + failed []failureWant + }{ + {"a plugin the store has not seen", func(*memoryStore, *versionedPlugin) {}, + []string{"versions", "alpha version", "alpha migrate", "record alpha 1", "alpha start"}, + []string{"alpha"}, nil}, + {"a migration the store records as current", func(s *memoryStore, _ *versionedPlugin) { + s.versions["alpha"] = "1" + }, []string{"versions", "alpha version", "alpha start"}, []string{"alpha"}, nil}, + {"a failed record, whose plugin still starts", func(s *memoryStore, _ *versionedPlugin) { + s.recordErr = errStoreOffline + }, []string{"versions", "alpha version", "alpha migrate", "record alpha 1", "alpha start"}, + []string{"alpha"}, []failureWant{ + {"alpha", pluginkit.StepRecord, false, "pluginkit: alpha record: store offline"}, + }}, + {"a panicking record, whose plugin still starts", func(s *memoryStore, _ *versionedPlugin) { + s.recordPanic = true + }, []string{"versions", "alpha version", "alpha migrate", "record alpha 1", "alpha start"}, + []string{"alpha"}, []failureWant{ + {"alpha", pluginkit.StepRecord, true, "pluginkit: alpha record panicked: boom"}, + }}, + {"a failed record and then a failed start", func(s *memoryStore, p *versionedPlugin) { + s.recordErr = errStoreOffline + p.startErr = errBoot + }, []string{"versions", "alpha version", "alpha migrate", "record alpha 1", "alpha start"}, + nil, []failureWant{ + {"alpha", pluginkit.StepRecord, false, "pluginkit: alpha record: store offline"}, + {"alpha", pluginkit.StepStart, false, "pluginkit: alpha start: boot failed"}, + }}, + {"a failed migration, which it does not record", func(_ *memoryStore, p *versionedPlugin) { + p.migrateErr = errSchema + }, []string{"versions", "alpha version", "alpha migrate"}, nil, []failureWant{ + {"alpha", pluginkit.StepMigrate, false, "pluginkit: alpha migrate: schema exploded"}, + }}, + {"an empty version", func(_ *memoryStore, p *versionedPlugin) { p.version = "" }, + []string{"versions", "alpha version"}, nil, []failureWant{ + {"alpha", pluginkit.StepMigrate, false, "pluginkit: alpha migration version: the version is empty"}, + }}, + {"a version the plugin cannot tell", func(_ *memoryStore, p *versionedPlugin) { p.versionErr = errUnreadable }, + []string{"versions", "alpha version"}, nil, []failureWant{ + {"alpha", pluginkit.StepMigrate, false, "pluginkit: alpha migration version: folder unreadable"}, + }}, + {"a version that panics", func(_ *memoryStore, p *versionedPlugin) { p.versionPanic = true }, + []string{"versions", "alpha version"}, nil, []failureWant{ + {"alpha", pluginkit.StepMigrate, true, "pluginkit: alpha migration version panicked: boom"}, + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + var calls []string + store := &memoryStore{versions: map[string]string{}, calls: &calls} + alpha := versioned("alpha", "1", &calls) + tc.prepare(store, alpha) + + outcome := startEach(t.Context(), t, storeHost(t, store, nil, alpha)) + + wantOutcome(t, outcome, tc.started, tc.failed) + wantCalls(t, calls, tc.calls) + }) + } +} + +// noMapStore is a migration store whose read answers a nil map. +type noMapStore struct { + memoryStore +} + +// Versions answers a nil map, as a store with no records yet may. +func (s *noMapStore) Versions(ctx context.Context) (map[string]string, error) { + _, err := s.memoryStore.Versions(ctx) + return nil, err +} + +func TestStartEachMigratesEveryPluginWhenTheStoreReadsNoMap(t *testing.T) { + t.Parallel() + + var calls []string + store := &noMapStore{memoryStore{versions: map[string]string{}, calls: &calls}} + host, err := pluginkit.NewHostWithMigrationStore(store, nil, versioned("alpha", "1", &calls)) + if err != nil { + t.Fatalf("NewHostWithMigrationStore() error = %v, want nil", err) + } + + outcome := startEach(t.Context(), t, host) + + wantOutcome(t, outcome, []string{"alpha"}, nil) + wantCalls(t, calls, []string{"versions", "alpha version", "alpha migrate", "record alpha 1", "alpha start"}) +} + +func TestStartEachUsesTheMigratorTheLookupGives(t *testing.T) { + t.Parallel() + + var calls []string + store := &memoryStore{versions: map[string]string{}, calls: &calls} + host := storeHost(t, store, lookupFor("alpha", "4", &calls), plain("alpha", &calls), migrating("beta", &calls)) + + outcome := startEach(t.Context(), t, host) + + wantOutcome(t, outcome, []string{"alpha", "beta"}, nil) + wantCalls(t, calls, []string{ + "versions", "alpha schema migrate", "record alpha 4", "beta migrate", "alpha start", "beta start", + }) +} + +func TestStartEachGoesOnPastABrokenLookup(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + lookup pluginkit.MigratorOf + panicked bool + text string + }{ + {"a lookup that panics", func(pluginkit.Plugin) (pluginkit.Migrator, bool) { panic("boom") }, + true, "pluginkit: alpha migrator lookup panicked: boom"}, + {"a lookup that gives a nil migrator", func(pluginkit.Plugin) (pluginkit.Migrator, bool) { return nil, true }, + false, "pluginkit: alpha migrator lookup: the lookup gave a nil migrator"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + var calls []string + store := &memoryStore{versions: map[string]string{}, calls: &calls} + lookup := func(p pluginkit.Plugin) (pluginkit.Migrator, bool) { + if p.ID() == "alpha" { + return tc.lookup(p) + } + return nil, false + } + host := storeHost(t, store, lookup, versioned("alpha", "1", &calls), plain("beta", &calls)) + + outcome := startEach(t.Context(), t, host) + + wantOutcome(t, outcome, []string{"beta"}, []failureWant{ + {"alpha", pluginkit.StepMigrate, tc.panicked, tc.text}, + }) + wantCalls(t, calls, []string{"versions", "beta start"}) + }) + } +} + +func TestStartEachWithoutAStoreNeverAsksAVersion(t *testing.T) { + t.Parallel() + + var calls []string + host := pluginkit.NewHost(versioned("alpha", "1", &calls)) + + outcome := startEach(t.Context(), t, host) + + wantOutcome(t, outcome, []string{"alpha"}, nil) + wantCalls(t, calls, []string{"alpha migrate", "alpha start"}) +} + +func TestStartEachReturnsAStoreReadFailureAndCallsNoPlugin(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + store func(calls *[]string) *memoryStore + text string + cause error + }{ + {"a read that fails", func(c *[]string) *memoryStore { + return &memoryStore{readErr: errStoreOffline, calls: c} + }, "pluginkit: read the migration versions: store offline", errStoreOffline}, + {"a read that panics", func(c *[]string) *memoryStore { + return &memoryStore{readPanic: true, calls: c} + }, "pluginkit: read the migration versions panicked: boom", nil}, + {"a read the context ended", func(c *[]string) *memoryStore { + return &memoryStore{readErr: context.Canceled, calls: c} + }, "pluginkit: read the migration versions: context canceled", context.Canceled}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + var calls []string + host := storeHost(t, tc.store(&calls), nil, versioned("alpha", "1", &calls), plain("beta", &calls)) + + outcome, err := host.StartEach(t.Context()) + + wantFailure(t, err, tc.text) + if tc.cause != nil && !errors.Is(err, tc.cause) { + t.Errorf("StartEach() error = %v, want %v in its chain", err, tc.cause) + } + wantOutcome(t, outcome, nil, nil) + wantCalls(t, calls, []string{"versions"}) + }) + } +} + +func TestHostStopAfterStartEachReachesEveryPlugin(t *testing.T) { + t.Parallel() + + var calls []string + unmigrated := migrating("gamma", &calls) + unmigrated.migrateErr = errors.New("schema exploded") + host := pluginkit.NewHost( + plain("alpha", &calls), + &fakePlugin{id: "beta", startErr: errors.New("boot failed"), calls: &calls}, + unmigrated, + ) + + startEach(t.Context(), t, host) + wantCalls(t, calls, []string{"gamma migrate", "alpha start", "beta start"}) + + if err := host.Stop(t.Context()); err != nil { + t.Fatalf("Stop() error = %v, want nil", err) + } + wantCalls(t, calls, []string{ + "gamma migrate", "alpha start", "beta start", "gamma stop", "beta stop", "alpha stop", + }) +} + +// slowPlugin is a plugin whose Migrate and Start each spend time. +type slowPlugin struct { + fakePlugin + migrateTakes time.Duration + startTakes time.Duration +} + +// Migrate spends the time the plugin was given. +func (s *slowPlugin) Migrate(context.Context) error { + time.Sleep(s.migrateTakes) + return nil +} + +// MigrationVersion returns a version no store has recorded. +func (s *slowPlugin) MigrationVersion() (string, error) { + return "1", nil +} + +// Start spends the time the plugin was given. +func (s *slowPlugin) Start(ctx context.Context) error { + time.Sleep(s.startTakes) + return s.fakePlugin.Start(ctx) +} + +func TestStartEachReportsHowLongAStartTookAfterAFailedRecord(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + var calls []string + store := &memoryStore{versions: map[string]string{}, recordErr: errStoreOffline, calls: &calls} + slow := &slowPlugin{fakePlugin: fakePlugin{id: "alpha", calls: &calls}, startTakes: 3 * time.Minute} + + outcome := startEach(t.Context(), t, storeHost(t, store, nil, slow)) + + want := []pluginkit.StartedPlugin{{ID: "alpha", Took: 3 * time.Minute}} + if !slices.Equal(want, outcome.Started) { + t.Errorf("started = %v, want %v", outcome.Started, want) + } + }) +} + +func TestStartEachReportsHowLongEachStartTook(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + var calls []string + host := pluginkit.NewHost( + &slowPlugin{ + fakePlugin: fakePlugin{id: "alpha", calls: &calls}, + migrateTakes: 2 * time.Minute, + startTakes: 3 * time.Minute, + }, + &slowPlugin{ + fakePlugin: fakePlugin{id: "beta", startErr: errors.New("boot failed"), calls: &calls}, + startTakes: 7 * time.Minute, + }, + &slowPlugin{fakePlugin: fakePlugin{id: "gamma", calls: &calls}, startTakes: 5 * time.Minute}, + plain("delta", &calls), + ) + + outcome := startEach(t.Context(), t, host) + + want := []pluginkit.StartedPlugin{ + {ID: "alpha", Took: 3 * time.Minute}, + {ID: "gamma", Took: 5 * time.Minute}, + {ID: "delta", Took: 0}, + } + if !slices.Equal(want, outcome.Started) { + t.Errorf("started = %v, want %v", outcome.Started, want) + } + }) +} diff --git a/pluginkit/store_test.go b/pluginkit/store_test.go index 83727f6..7f365fe 100644 --- a/pluginkit/store_test.go +++ b/pluginkit/store_test.go @@ -363,6 +363,20 @@ func TestHostWithMigrationStoreRecoversALookupPanicNamingThePlugin(t *testing.T) wantCalls(t, calls, []string{"versions"}) } +func TestHostWithMigrationStoreRefusesANilMigratorNamingThePlugin(t *testing.T) { + t.Parallel() + + var calls []string + store := &memoryStore{versions: map[string]string{}, calls: &calls} + lookup := func(pluginkit.Plugin) (pluginkit.Migrator, bool) { return nil, true } + host := storeHost(t, store, lookup, plain("alpha", &calls)) + + err := host.Migrate(t.Context()) + + wantFailure(t, err, "pluginkit: alpha migrator lookup: the lookup gave a nil migrator") + wantCalls(t, calls, []string{"versions"}) +} + // contextStore is a migration store that adds the caller value of each context it gets to seen. type contextStore struct { seen *[]any From eebe0727701686aa36eada44414d1f11f21dd52e Mon Sep 17 00:00:00 2001 From: SirLouen Date: Tue, 3 Nov 2026 23:45:05 +0100 Subject: [PATCH 3/3] docs(pluginkit): list the Migrate, Routes and PublicPaths doc changes in the changelog --- pluginkit/CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/pluginkit/CHANGELOG.md b/pluginkit/CHANGELOG.md index 91bec6a..a781da0 100644 --- a/pluginkit/CHANGELOG.md +++ b/pluginkit/CHANGELOG.md @@ -10,6 +10,7 @@ Releases of this module are tagged `pluginkit/vX.Y.Z`. Releases up to - An optional `MigratorOf` lookup gives that host the `Migrator` of a plugin that ships its schema as data. - `Host.StartEach` migrates and then starts every plugin on its own, goes on past a failure or a panic, and answers a `StartOutcome` naming each plugin that started and each `Failure`. - The `Plugin` doc promises that `Stop` may run more than once and that `Start` may run again after it, and asks `Start` to return by the time its context ends. +- The `Migrator` doc asks `Migrate` to return by the time its context ends, and the `RouteProvider` and `PublicPathProvider` docs ask `Routes` and `PublicPaths` to work whether or not `Start` ran or worked. ## 0.7.0 - 2026-10-15