diff --git a/core/services/addons/manager.go b/core/services/addons/manager.go index cffceb848d..4c0e10c5a1 100644 --- a/core/services/addons/manager.go +++ b/core/services/addons/manager.go @@ -102,6 +102,7 @@ type Manager struct { mu sync.Mutex upgradeMu sync.Mutex upgradeViews map[string]upgradeSnapshot + finishedUpgrades []string // oldest first; guarded by upgradeMu upgradeCtx context.Context upgradeCancel context.CancelFunc upgradeWG sync.WaitGroup @@ -260,7 +261,7 @@ func NewManager(ctx context.Context, cfg OwnerConfig, c catalog.Catalog, i *pack _ = release() return nil, e } - m.refreshUpgradeViews() + m.syncUpgradeViews(true) return m, nil } func (m *Manager) Close() error { @@ -465,7 +466,19 @@ func (m *Manager) submit(ctx context.Context, actor contract.Actor, cmd Lifecycl return OperationStatus{}, ErrConflict } if accepted { - phase, _ := upgradeTransition(lifecyclePhase(op.Status.Phase), upgradeStart) + phase, _, err := upgradeTransition(lifecyclePhase(op.Status.Phase), upgradeStart) + if err != nil { + // A phase the reducer does not know cannot be run. End the + // operation as failed instead of leaving it half-owned. + xlog.Warn("Addon operation has an unknown phase", "operationId", cmd.OperationID, "phase", op.Status.Phase) + op.Status.Phase = string(lifecycleFailed) + op.Status.FailureCode = "upgrade_failed" + m.journal.Operations[cmd.OperationID] = op + if e := m.persist(string(lifecycleFailed)); e != nil { + return op.Status, e + } + return op.Status, ErrOperationFailed + } op.Status.Phase = string(phase) m.journal.Operations[cmd.OperationID] = op if e := m.persist(string(lifecycleRunning)); e != nil { diff --git a/core/services/addons/upgrade_operation.go b/core/services/addons/upgrade_operation.go index 05cb5c0990..02dfde6a98 100644 --- a/core/services/addons/upgrade_operation.go +++ b/core/services/addons/upgrade_operation.go @@ -6,10 +6,14 @@ import ( "errors" "github.com/mudler/LocalAI/pkg/addons/contract" "github.com/mudler/xlog" + "slices" ) type upgradeExecutionKey struct{} +// maxFinishedUpgradeViews bounds the views kept for finished upgrades. +const maxFinishedUpgradeViews = 256 + type upgradeSnapshot struct { Command LifecycleCommand Actor contract.Actor @@ -99,8 +103,7 @@ func (m *Manager) SubmitUpgrade(ctx context.Context, actor contract.Actor, cmd L m.mu.Lock() current := m.journal.Operations[cmd.OperationID] if err != nil && current.Status.Phase != string(lifecycleCompleted) && current.Status.Phase != string(lifecycleFailed) && !m.uncertain { - phase, _ := upgradeTransition(lifecycleRunning, upgradeFail) - current.Status.Phase = string(phase) + current.Status.Phase = string(lifecycleFailed) current.Status.FailureCode = "upgrade_failed" if errors.Is(err, context.Canceled) { current.Status.FailureCode = "interrupted" @@ -124,17 +127,39 @@ func (m *Manager) SubmitUpgrade(ctx context.Context, actor contract.Actor, cmd L // refreshUpgradeViews is called only while mu is held. Values contain no // mutable slices or pointers and may be returned without the execution lock. -func (m *Manager) refreshUpgradeViews() { +// +// Finished operations keep a view only while they are among the newest +// maxFinishedUpgradeViews; the journal still answers for older ones. +func (m *Manager) refreshUpgradeViews() { m.syncUpgradeViews(false) } + +// syncUpgradeViews projects the journal. At start (loadFinished) the finished +// operations of earlier processes get a view too, so a status read keeps its +// failure code; afterwards only operations that already have one do. +func (m *Manager) syncUpgradeViews(loadFinished bool) { m.upgradeMu.Lock() defer m.upgradeMu.Unlock() for id, op := range m.journal.Operations { - if op.Async { - status := op.Status - if m.uncertain || m.runtimeRecoveryRequired { - status.RuntimeState = string(lifecycleRecovery) - } - m.upgradeViews[id] = upgradeSnapshot{op.Command, op.Actor, status} + if !op.Async { + continue } + _, viewed := m.upgradeViews[id] + if !isActive(op.Status.Phase) { + if !viewed && !loadFinished { + continue + } + if !slices.Contains(m.finishedUpgrades, id) { + m.finishedUpgrades = append(m.finishedUpgrades, id) + } + } + status := op.Status + if m.uncertain || m.runtimeRecoveryRequired { + status.RuntimeState = string(lifecycleRecovery) + } + m.upgradeViews[id] = upgradeSnapshot{op.Command, op.Actor, status} + } + for len(m.finishedUpgrades) > maxFinishedUpgradeViews { + delete(m.upgradeViews, m.finishedUpgrades[0]) + m.finishedUpgrades = m.finishedUpgrades[1:] } } diff --git a/core/services/addons/upgrade_state.go b/core/services/addons/upgrade_state.go index bb7bff7637..0f59fb7fab 100644 --- a/core/services/addons/upgrade_state.go +++ b/core/services/addons/upgrade_state.go @@ -1,6 +1,13 @@ // SPDX-License-Identifier: MIT package addons +import "errors" + +// errInvalidUpgradeTransition means the phase or the event is not one the +// reducer knows. The phase can come from a journal on disk, so it is an error +// for the caller to handle, not a reason to crash a worker. +var errInvalidUpgradeTransition = errors.New("invalid upgrade state or event") + // upgradeTransition is the owner's observation reducer. Effects remain in the // existing lifecycle and recovery journals; observer detach never emits work. // Every terminal result is sticky, including after late progress or a retry. @@ -16,25 +23,31 @@ const ( upgradeRetry upgradeEvent = "retry" ) -func upgradeTransition(phase lifecyclePhase, event upgradeEvent) (next lifecyclePhase, rule string) { +func upgradeTransition(phase lifecyclePhase, event upgradeEvent) (next lifecyclePhase, rule string, err error) { switch phase { case lifecycleCompleted, lifecycleFailed, lifecycleRecovery: - return phase, "U0-terminal" + return phase, "U0-terminal", nil case lifecycleAccepted, lifecycleRunning: switch event { case upgradeDetach, upgradeRetry: - return phase, "U1-observe" + return phase, "U1-observe", nil case upgradeStart: - return lifecycleRunning, "U2-start" + return lifecycleRunning, "U2-start", nil case upgradeComplete: - return lifecycleCompleted, "U3-complete" + return lifecycleCompleted, "U3-complete", nil case upgradeFail: - return lifecycleFailed, "U4-fail" + return lifecycleFailed, "U4-fail", nil case upgradeInterrupt: - return lifecycleFailed, "U5-interrupted" + return lifecycleFailed, "U5-interrupted", nil case upgradeRecover: - return lifecycleRecovery, "U6-recovery" + return lifecycleRecovery, "U6-recovery", nil } } - panic("invalid upgrade state or event") + return phase, "", errInvalidUpgradeTransition +} + +// isActive reports whether an operation in this phase still blocks its add-on: +// anything but completed or failed, including recovery-required. +func isActive(phase string) bool { + return phase != string(lifecycleCompleted) && phase != string(lifecycleFailed) } diff --git a/core/services/addons/upgrade_state_test.go b/core/services/addons/upgrade_state_test.go index f649368eb8..7cb5004bb3 100644 --- a/core/services/addons/upgrade_state_test.go +++ b/core/services/addons/upgrade_state_test.go @@ -2,6 +2,8 @@ package addons import ( + "fmt" + . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" ) @@ -22,7 +24,7 @@ var _ = Describe("upgrade observation reducer", func() { phase := lifecycleAccepted for n, event := range trace.events { var rule string - phase, rule = upgradeTransition(phase, event) + phase, rule, _ = upgradeTransition(phase, event) Expect(rule).To(Equal(trace.rules[n])) } Expect(phase).To(Equal(trace.end)) @@ -32,21 +34,49 @@ var _ = Describe("upgrade observation reducer", func() { events := []upgradeEvent{upgradeStart, upgradeRetry, upgradeDetach, upgradeComplete, upgradeFail, upgradeInterrupt, upgradeRecover} for _, phase := range []lifecyclePhase{lifecycleAccepted, lifecycleRunning, lifecycleCompleted, lifecycleFailed, lifecycleRecovery} { for _, event := range []upgradeEvent{upgradeStart, upgradeComplete, upgradeFail, upgradeInterrupt, upgradeRecover, upgradeDetach, upgradeRetry} { - next, rule := upgradeTransition(phase, event) + next, rule, _ := upgradeTransition(phase, event) Expect(next).NotTo(BeEmpty()) Expect(rule).NotTo(BeEmpty()) } } for _, phase := range []lifecyclePhase{lifecycleCompleted, lifecycleFailed, lifecycleRecovery} { for _, event := range events { - next, _ := upgradeTransition(phase, event) + next, _, _ := upgradeTransition(phase, event) Expect(next).To(Equal(phase)) } } for _, phase := range []lifecyclePhase{lifecycleAccepted, lifecycleRunning} { - next, rule := upgradeTransition(phase, upgradeInterrupt) + next, rule, _ := upgradeTransition(phase, upgradeInterrupt) Expect(next).To(Equal(lifecycleFailed)) Expect(rule).To(Equal("U5-interrupted")) } }) + It("refuses an unknown phase or event with an error instead of panicking", func() { + for _, phase := range []lifecyclePhase{"", "garbage", lifecycleIntent} { + next, _, err := upgradeTransition(phase, upgradeStart) + Expect(err).To(MatchError(errInvalidUpgradeTransition)) + Expect(next).To(Equal(phase)) + } + _, _, err := upgradeTransition(lifecycleAccepted, upgradeEvent("bogus")) + Expect(err).To(MatchError(errInvalidUpgradeTransition)) + }) + It("drops the oldest finished views once too many pile up", func() { + m := &Manager{upgradeViews: map[string]upgradeSnapshot{}, journal: journal{Operations: map[string]journalOperation{}}} + for i := 0; i < maxFinishedUpgradeViews+10; i++ { + id := fmt.Sprintf("op-%04d", i) + m.journal.Operations[id] = journalOperation{Async: true, Status: OperationStatus{OperationID: id, Phase: string(lifecycleAccepted)}} + m.refreshUpgradeViews() + op := m.journal.Operations[id] + op.Status.Phase = string(lifecycleCompleted) + m.journal.Operations[id] = op + m.refreshUpgradeViews() + } + m.journal.Operations["live"] = journalOperation{Async: true, Status: OperationStatus{OperationID: "live", Phase: string(lifecycleRunning)}} + m.refreshUpgradeViews() + Expect(m.upgradeViews).To(HaveLen(maxFinishedUpgradeViews + 1)) + Expect(m.upgradeViews).To(HaveKey("live")) + Expect(m.upgradeViews).NotTo(HaveKey("op-0000")) + Expect(m.upgradeViews).To(HaveKey(fmt.Sprintf("op-%04d", maxFinishedUpgradeViews+9))) + + }) })