From 10722d8e3bbbd1dacde65e27069813256cdc1c29 Mon Sep 17 00:00:00 2001 From: mudler-agent Date: Mon, 2 Nov 2026 22:47:17 +0100 Subject: [PATCH] fix(addons): run accepted upgrades on the owner (#13035) * fix(addons): run accepted upgrades on the owner HTTP cancellation during a verified backup can prevent package commit. Accept browser upgrades durably before replying and execute them under the owner lifecycle, with bounded admission and same-operation polling. Keep synchronous lifecycle callers unchanged. Reuse the publication and data recovery journals rather than replay interrupted migration. Expose real stages and safe terminal causes, and recover browser observers after reload without resubmitting work. Assisted-by: nib:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * fix(addons): isolate upgrade observers by session Reset action identities when the principal changes and restore pending identities before polling. A different account must never submit an ID owned by the previous account. Clear stored upgrade observers directly in logout before navigation. Cleanup also handles a fresh store with no bound principal. Assisted-by: nib:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * refactor(addons): use lifecycle phase constants Share named lifecycle phases between the journal and upgrade reducer. Use typed events in transitions and trace tests instead of repeating wire literals. Keep operation DTOs, serialized values and rule labels unchanged. Assisted-by: nib:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * fix(addons): verify accepted upgrades to completion Preserve typed pin refusals before accepting an upgrade and report the same transition conflict while execution holds the manager lock. Poll accepted operations in HTTP and real-process acceptance tests before asserting publication or failure. Exercise reload and truthful stages in the rendered Activity test, retaining version details without cancellation controls or estimated percentages. Register the new verification cases. Assisted-by: nib:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * test(addons): await the Tables upgrade receipt An accepted upgrade can still serve its previous generation. Poll the same operation through another frontend until it reaches a terminal state before checking the new version, retained data, grants and backup. Verify that a repeated submission returns the same completed receipt. Assisted-by: nib:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * test(addons): poll the winning upgrade operation Concurrent submissions return acceptance before publication. Observe the winning browser session and operation until completion before asserting the installed generation. Keep the losing transition conflict check and require the winning ID to replay the same terminal receipt. Assisted-by: nib:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * test(ui): model authoritative owner settings Host controls now read and patch revisioned owner policy through dispatch. Update rendered fixtures to serve that strict envelope rather than local runtime settings. Keep private confirmation browser-only, send per-key provider changes, and keep node-local audit retention outside owner forms. Preserve the skill default in the tools fixture so an unpinned choice matches the provider the owner reports. Assisted-by: nib:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * test(ui): assert the owner policy write envelope Verify that the private-network switch sends no owner write before its confirmation phrase. After confirmation, assert the actual dispatch action, operation identity, expected revision and policy-only patch. Assisted-by: nib:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto * test(addons): cover shutdown and lifecycle exclusion Exercise owner shutdown during the real backup path and verify Close joins cleanup before releasing storage. Cover synchronous disable while an upgrade is accepted, including the journal guard before reconciliation. Document that accepted upgrades have no UI cancellation control. Register the new lifecycle cases in the canonical suite inventory. Assisted-by: nib:gpt-6.1-sol Signed-off-by: Ettore Di Giacinto --------- Signed-off-by: Ettore Di Giacinto Co-authored-by: Ettore Di Giacinto --- core/http/endpoints/addons/catalogui_test.go | 4 +- core/http/endpoints/addons/updates_test.go | 29 +- core/http/react-ui/e2e/addons-upgrade.spec.js | 22 +- core/http/react-ui/e2e/chats-egress.spec.js | 5 + core/http/react-ui/e2e/settings-tools.spec.js | 2 +- core/http/react-ui/e2e/update-review.spec.js | 36 ++- .../react-ui/public/locales/de/operate.json | 22 +- .../react-ui/public/locales/en/operate.json | 22 +- .../react-ui/public/locales/es/operate.json | 22 +- .../react-ui/public/locales/fr/operate.json | 22 +- .../react-ui/public/locales/id/operate.json | 22 +- .../react-ui/public/locales/it/operate.json | 22 +- .../react-ui/public/locales/ko/operate.json | 22 +- .../public/locales/pt-BR/operate.json | 22 +- .../public/locales/zh-CN/operate.json | 22 +- .../react-ui/src/components/OperationCard.jsx | 5 +- .../src/components/addons/actionBook.js | 4 + .../http/react-ui/src/context/AuthContext.jsx | 6 + .../src/contexts/OperationsContext.jsx | 9 +- core/http/react-ui/src/utils/addonUpgrades.js | 59 +++- core/http/react-ui/src/utils/addonsApi.js | 2 + .../react-ui/tests/unit/updatereview.test.mjs | 79 ++++- .../http/react-ui/tests/unit/updates.test.mjs | 2 +- core/services/addons/gateway/gateway.go | 8 +- core/services/addons/gateway/updates.go | 2 +- .../addons/gateway/upgrade_polling_test.go | 97 ++++++ core/services/addons/journal.go | 6 + core/services/addons/manager.go | 130 ++++++-- core/services/addons/recovery.go | 8 +- core/services/addons/recovery/backup.go | 12 +- core/services/addons/recovery/migration.go | 13 +- core/services/addons/state.go | 17 + core/services/addons/upgrade_operation.go | 149 +++++++++ .../services/addons/upgrade_operation_test.go | 290 ++++++++++++++++++ core/services/addons/upgrade_state.go | 40 +++ core/services/addons/upgrade_state_test.go | 52 ++++ docs/content/development/addon-testing.md | 19 ++ docs/content/features/addons.md | 38 +++ scripts/addons/suites.json | 42 ++- .../foundation/recovery_failure_test.go | 23 +- .../addons/tablesacceptance/replicas_test.go | 18 ++ 41 files changed, 1324 insertions(+), 102 deletions(-) create mode 100644 core/services/addons/gateway/upgrade_polling_test.go create mode 100644 core/services/addons/upgrade_operation.go create mode 100644 core/services/addons/upgrade_operation_test.go create mode 100644 core/services/addons/upgrade_state.go create mode 100644 core/services/addons/upgrade_state_test.go diff --git a/core/http/endpoints/addons/catalogui_test.go b/core/http/endpoints/addons/catalogui_test.go index c9a4b7243..dba852f2d 100644 --- a/core/http/endpoints/addons/catalogui_test.go +++ b/core/http/endpoints/addons/catalogui_test.go @@ -261,7 +261,8 @@ var _ = Describe("catalog and management actions", Label("catalogui"), func() { generation := uint64(before["generation"].(float64)) r := send(e, "POST", "/api/addons/dispatch", dispatchBody("upgrade-"+name, "headless", "upgrade", 0, `{"sourceId":"test","version":"1.0.0"}`), admin) Expect(r.code).To(Equal(200), name) - Expect(payloadOf(r)).To(HaveKeyWithValue("Phase", "completed")) + Expect(payloadOf(r)).To(HaveKeyWithValue("Phase", "accepted")) + Expect(awaitLifecycle(e, "headless", "upgrade-"+name, admin)).To(HaveKeyWithValue("Phase", "completed")) after := byID(listOf(payloadOf(send(e, "POST", "/api/addons/dispatch", dispatchBody("up-after-"+name, "", "addons-installed", 0, `{}`), admin)), "addons"), "id", "headless") Expect(uint64(after["generation"].(float64))).To(Equal(generation + 1)) Expect(after).To(HaveKeyWithValue("generations", BeNumerically("==", generation+1))) @@ -271,6 +272,7 @@ var _ = Describe("catalog and management actions", Label("catalogui"), func() { // The same operation replays and never upgrades twice. r := send(s.frontends["owner"], "POST", "/api/addons/dispatch", dispatchBody("upgrade-replay", "headless", "upgrade", 0, `{"sourceId":"test","version":"1.0.0"}`), bearerOf(s.tokens["alice"])) Expect(r.code).To(Equal(200)) + Expect(awaitLifecycle(s.frontends["owner"], "headless", "upgrade-replay", bearerOf(s.tokens["alice"]))).To(HaveKeyWithValue("Phase", "completed")) first, err := s.manager.Status(context.Background(), "headless") Expect(err).NotTo(HaveOccurred()) again := send(s.frontends["front-a"], "POST", "/api/addons/dispatch", dispatchBody("upgrade-replay", "headless", "upgrade", 0, `{"sourceId":"test","version":"1.0.0"}`), bearerOf(s.tokens["alice"])) diff --git a/core/http/endpoints/addons/updates_test.go b/core/http/endpoints/addons/updates_test.go index 66044fce8..7e32b9703 100644 --- a/core/http/endpoints/addons/updates_test.go +++ b/core/http/endpoints/addons/updates_test.go @@ -160,6 +160,20 @@ func (s *updatesStack) generation() uint64 { return st.Generation } +// awaitLifecycle reads through the authenticated HTTP surface, never treating +// an acceptance reply as proof that a generation was published. +func awaitLifecycle(e http.Handler, addon, op string, headers func(*http.Request)) map[string]any { + GinkgoHelper() + var status map[string]any + Eventually(func() string { + r := send(e, "POST", "/api/addons/dispatch", dispatchBody(op, addon, "operation-status", 0, `{}`), headers) + Expect(r.code).To(Equal(200), fmt.Sprint(r.body)) + status = payloadOf(r) + return status["Phase"].(string) + }, 5*time.Second).Should(Or(Equal("completed"), Equal("failed"))) + return status +} + var _ = Describe("update routes", Label("updates"), func() { var s *updatesStack BeforeEach(func() { s = newUpdatesStack() }) @@ -246,13 +260,15 @@ var _ = Describe("update routes", Label("updates"), func() { done := send(s.frontends["front-a"], "POST", path, upgradeBody("up-1", before, next.Digest), alice("up-1")) Expect(done.code).To(Equal(200), fmt.Sprint(done.body)) - Expect(done.body["payload"]).To(HaveKeyWithValue("Phase", "completed")) + Expect(done.body["payload"]).To(HaveKeyWithValue("Phase", "accepted")) + terminal := awaitLifecycle(s.frontends["front-b"], "notes", "up-1", alice("up-1")) + Expect(terminal).To(HaveKeyWithValue("Phase", "completed")) Expect(s.generation()).To(Equal(uint64(2))) Expect(s.installedDigest()).To(Equal(next.Digest)) // The same operation, from another frontend, replays; it runs nowhere again. replay := send(s.frontends["front-b"], "POST", path, upgradeBody("up-1", before, next.Digest), alice("up-1")) Expect(replay.code).To(Equal(200)) - Expect(replay.body["payload"]).To(Equal(done.body["payload"])) + Expect(replay.body["payload"]).To(Equal(terminal)) // Another operation ID for the same transition is a conflict. other := send(s.frontends["owner"], "POST", path, upgradeBody("up-2", before, next.Digest), alice("up-2")) Expect(other.code).To(Equal(409)) @@ -285,7 +301,9 @@ var _ = Describe("update routes", Label("updates"), func() { same := race([2]string{"same", "same"}, before, next.Digest) Expect(same[0].code).To(Equal(200)) Expect(same[1].code).To(Equal(200)) - Expect(same[0].body["payload"]).To(Equal(same[1].body["payload"])) + Expect(same[0].body["payload"]).To(HaveKeyWithValue("OperationID", "same")) + Expect(same[1].body["payload"]).To(HaveKeyWithValue("OperationID", "same")) + Expect(awaitLifecycle(s.frontends["owner"], "notes", "same", alice("same"))).To(HaveKeyWithValue("Phase", "completed")) Expect(s.generation()).To(Equal(uint64(2))) // A new candidate, and two different operation IDs for it. @@ -301,6 +319,11 @@ var _ = Describe("update routes", Label("updates"), func() { Expect(r.errType()).To(Equal("upgrade_conflict")) } } + for n, r := range differ { + if r.code == 200 { + Expect(awaitLifecycle(s.frontends["owner"], "notes", []string{"first", "second"}[n], alice([]string{"first", "second"}[n]))).To(HaveKeyWithValue("Phase", "completed")) + } + } Expect(s.generation()).To(Equal(uint64(3))) }) diff --git a/core/http/react-ui/e2e/addons-upgrade.spec.js b/core/http/react-ui/e2e/addons-upgrade.spec.js index e9d30be8c..57ec01032 100644 --- a/core/http/react-ui/e2e/addons-upgrade.spec.js +++ b/core/http/react-ui/e2e/addons-upgrade.spec.js @@ -238,13 +238,33 @@ test.describe('Add-on updates and the explicit upgrade through the real owner', expect(results.map(r => r.status).sort(), JSON.stringify(results)).toEqual([200, 409]) const loser = results.find(r => r.status === 409) expect(loser.json.error.type).toBe('upgrade_conflict') + const winnerIndex = results.findIndex(r => r.status === 200) + const winnerId = [one, two][winnerIndex] + const winnerPage = [a.page, b.page][winnerIndex] + expect(results[winnerIndex].json.payload.OperationID).toBe(winnerId) + expect(results[winnerIndex].json.payload.Phase).toBe('accepted') + // Acceptance does not publish a generation. Poll as the winning browser + // session so this read remains bound to the actor that submitted the work. + let terminal + await expect.poll(async () => { + const status = await pageFetch(winnerPage, '/api/addons/dispatch', { + method: 'POST', body: { addonId: 'tables', operationId: winnerId, action: 'operation-status', arguments: {} }, + }) + expect(status.status, JSON.stringify(status.json)).toBe(200) + terminal = status.json.payload + expect(terminal.OperationID).toBe(winnerId) + return terminal.Phase + }).toMatch(/^(completed|failed)$/) + expect(terminal.Phase, JSON.stringify(terminal)).toBe('completed') + expect(terminal.Generation).toBe(3) const after = await installed(api) expect(after.generation).toBe(3) expect(after.digest).toBe(now.update.digest) state.thirdDigest = after.digest // The winner's operation replays and does not run again. - const again = await pageFetch(a.page, '/api/addons/tables/upgrade', { method: 'POST', body: body(results[0].status === 200 ? one : two), headers: { 'X-LocalAI-Confirm': results[0].status === 200 ? one : two } }) + const again = await pageFetch(winnerPage, '/api/addons/tables/upgrade', { method: 'POST', body: body(winnerId), headers: { 'X-LocalAI-Confirm': winnerId } }) expect(again.status).toBe(200) + expect(again.json.payload).toEqual(terminal) expect((await installed(api)).generation).toBe(3) await a.context.close() await b.context.close() diff --git a/core/http/react-ui/e2e/chats-egress.spec.js b/core/http/react-ui/e2e/chats-egress.spec.js index dc46d2abb..ac2116761 100644 --- a/core/http/react-ui/e2e/chats-egress.spec.js +++ b/core/http/react-ui/e2e/chats-egress.spec.js @@ -357,10 +357,15 @@ test.describe('The network of the commands on the add-on page', () => { await page.getByTestId('settings-egress-private-switch').click() await expect(page.getByRole('button', { name: 'Apply' })).toBeDisabled() + expect(settings.log.dispatch.filter(r => r.action === 'host-settings-update')).toHaveLength(0) await page.getByTestId('settings-egress-private-phrase').fill('I understand that commands can reach my private network') await page.getByRole('button', { name: 'Apply' }).click() // The phrase is checked in the page; the owner patch carries policy fields only. await expect.poll(() => settings.log.hostSettings).toEqual([{ chats_egress_allow_private: true }]) + const writes = settings.log.dispatch.filter(r => r.action === 'host-settings-update') + expect(writes).toHaveLength(1) + expect(writes[0].operationId).toBeTruthy() + expect(writes[0].arguments).toEqual({ expectedRevision: 1, patch: { chats_egress_allow_private: true } }) await expect(page.getByTestId('settings-pending')).toHaveCount(0) }) diff --git a/core/http/react-ui/e2e/settings-tools.spec.js b/core/http/react-ui/e2e/settings-tools.spec.js index 0522763c6..c09571106 100644 --- a/core/http/react-ui/e2e/settings-tools.spec.js +++ b/core/http/react-ui/e2e/settings-tools.spec.js @@ -179,7 +179,7 @@ test.describe('The skills of an add-on in its details', () => { }) await mockSettings(page) let rows = [ - { id: 'xlsx', provider: 'sheets', candidates: ['office', 'sheets'], titles: { office: 'Office', sheets: 'Sheets' } }, + { id: 'xlsx', provider: 'sheets', default: 'sheets', candidates: ['office', 'sheets'], titles: { office: 'Office', sheets: 'Sheets' } }, { id: 'docx', provider: 'office', candidates: ['office'], titles: { office: 'Office' } }, ] await page.route('**/api/chats/admin/tools', route => route.fulfill({ json: { tools: [], skills: rows } })) diff --git a/core/http/react-ui/e2e/update-review.spec.js b/core/http/react-ui/e2e/update-review.spec.js index 6aefd0a1b..c9881fe0c 100644 --- a/core/http/react-ui/e2e/update-review.spec.js +++ b/core/http/react-ui/e2e/update-review.spec.js @@ -47,20 +47,23 @@ test.describe('Add-on updates: after the confirmation', () => { expect(state.log.upgrade).toHaveLength(0) }) - test('an add-on upgrade is listed in Activity while it runs, can be canceled, and is recorded when it ends', async ({ page }) => { + test('an add-on upgrade is listed in Activity while it runs, survives reload, and is recorded when it ends', async ({ page }) => { await signIn(page) await page.route('**/api/operations', route => route.fulfill({ json: { operations: [] } })) await page.route('**/api/operations/history', route => route.fulfill({ json: { operations: [] } })) - let release - const gate = new Promise(resolve => { release = resolve }) - const state = await mockAddons(page, twoPending()) - // The owner answers an upgrade only when the spec lets it. - await page.route(/\/api\/addons\/wiki\/upgrade$/, async route => { - await gate - const body = JSON.parse(route.request().postData() || '{}') - const a = state.installed.find(x => x.id === 'wiki') - a.generation += 1; a.version = a.update.version; delete a.update - return route.fulfill(ok({ OperationID: body.operationId, AddonID: 'wiki', Phase: 'completed', Generation: a.generation })) + let finished = false + let operationId + const state = await mockAddons(page, twoPending({ + dispatch: body => { + if (body.action !== 'operation-status') return null + expect(body.operationId).toBe(operationId) + return ok({ OperationID: operationId, AddonID: 'wiki', Phase: finished ? 'completed' : 'running', stage: 'backing-up', Generation: finished ? 2 : 0 }) + }, + })) + await page.route(/\/api\/addons\/wiki\/upgrade$/, route => { + operationId = JSON.parse(route.request().postData() || '{}').operationId + state.log.upgrade.push({ operationId }) + return route.fulfill(ok({ OperationID: operationId, AddonID: 'wiki', Phase: 'accepted', Generation: 0 })) }) await page.goto('/app/addons') await page.locator('[data-testid="addon-tile"][data-app="wiki"]').getByTestId('update-upgrade-card').click() @@ -73,7 +76,12 @@ test.describe('Add-on updates: after the confirmation', () => { await expect(card).toBeVisible() await expect(card).toContainText('Upgrading add-on') await expect(card).toContainText('0.9.0 to 1.0.0') - await expect(card.getByRole('button', { name: /cancel/i })).toBeVisible() + await expect(card.getByRole('button', { name: /cancel/i })).toHaveCount(0) + await expect(card.getByRole('progressbar')).toHaveCount(0) + await expect(card).toContainText('Creating and verifying backup') + await page.reload() + await expect(card).toContainText('Creating and verifying backup') + expect(state.log.upgrade).toHaveLength(1) await expect(card.getByRole('button', { name: /pause/i })).toHaveCount(0) await page.getByRole('button', { name: 'Add-ons', exact: true }).click() @@ -82,7 +90,9 @@ test.describe('Add-on updates: after the confirmation', () => { await expect(card).toHaveCount(0) await page.getByRole('button', { name: 'Add-ons', exact: true }).click() - release() + const installed = state.installed.find(x => x.id === 'wiki') + installed.generation += 1; installed.version = installed.update.version; delete installed.update + finished = true await expect(card).toHaveCount(0) const record = page.locator('.activity-row').filter({ hasText: 'Wiki' }) await expect(record).toContainText('Upgraded') diff --git a/core/http/react-ui/public/locales/de/operate.json b/core/http/react-ui/public/locales/de/operate.json index 43c260d8d..46e9aa4b5 100644 --- a/core/http/react-ui/public/locales/de/operate.json +++ b/core/http/react-ui/public/locales/de/operate.json @@ -52,6 +52,26 @@ "rowUpgraded": "Aktualisiert in {{duration}}", "rowUpgradedPlain": "Aktualisiert", "viewInAddons": "In Add-ons anzeigen", - "upgradeWaiting": "Wartet auf den Start des Add-on-Dienstes. Nächste Prüfung in {{seconds}} s." + "upgradeWaiting": "Wartet auf den Start des Add-on-Dienstes. Nächste Prüfung in {{seconds}} s.", + "addonStage": { + "submitting": "Submitting upgrade", + "accepted": "Accepted by owner", + "running": "Running upgrade", + "preparing-package": "Preparing package", + "committing-package": "Committing package", + "backing-up": "Creating and verifying backup", + "backed-up": "Backup verified", + "staged": "Package committed", + "migrating": "Starting candidate and checking migration", + "healthy": "Candidate healthy", + "publishing": "Publishing generation", + "completed": "Upgrade completed", + "failed": "Upgrade failed" + }, + "addonFailure": { + "upgrade_failed": "Upgrade failed. Check the owner logs for details.", + "interrupted": "Upgrade interrupted by owner shutdown or restart. It was not replayed.", + "recovery_required": "Upgrade interrupted. Data recovery is required before this add-on can run." + } } } diff --git a/core/http/react-ui/public/locales/en/operate.json b/core/http/react-ui/public/locales/en/operate.json index 01715e855..ee5b74147 100644 --- a/core/http/react-ui/public/locales/en/operate.json +++ b/core/http/react-ui/public/locales/en/operate.json @@ -280,7 +280,27 @@ "rowUpgraded": "Upgraded in {{duration}}", "rowUpgradedPlain": "Upgraded", "viewInAddons": "View in Add-ons", - "upgradeWaiting": "Waiting for the add-on service to start. Next check in {{seconds}} s." + "upgradeWaiting": "Waiting for the add-on service to start. Next check in {{seconds}} s.", + "addonStage": { + "submitting": "Submitting upgrade", + "accepted": "Accepted by owner", + "running": "Running upgrade", + "preparing-package": "Preparing package", + "committing-package": "Committing package", + "backing-up": "Creating and verifying backup", + "backed-up": "Backup verified", + "staged": "Package committed", + "migrating": "Starting candidate and checking migration", + "healthy": "Candidate healthy", + "publishing": "Publishing generation", + "completed": "Upgrade completed", + "failed": "Upgrade failed" + }, + "addonFailure": { + "upgrade_failed": "Upgrade failed. Check the owner logs for details.", + "interrupted": "Upgrade interrupted by owner shutdown or restart. It was not replayed.", + "recovery_required": "Upgrade interrupted. Data recovery is required before this add-on can run." + } }, "backends": { "title": "Backend Management", diff --git a/core/http/react-ui/public/locales/es/operate.json b/core/http/react-ui/public/locales/es/operate.json index 06a9de206..8af8c5110 100644 --- a/core/http/react-ui/public/locales/es/operate.json +++ b/core/http/react-ui/public/locales/es/operate.json @@ -52,6 +52,26 @@ "rowUpgraded": "Actualizado en {{duration}}", "rowUpgradedPlain": "Actualizado", "viewInAddons": "Ver en Complementos", - "upgradeWaiting": "Esperando a que se inicie el servicio de complementos. Próxima comprobación en {{seconds}} s." + "upgradeWaiting": "Esperando a que se inicie el servicio de complementos. Próxima comprobación en {{seconds}} s.", + "addonStage": { + "submitting": "Submitting upgrade", + "accepted": "Accepted by owner", + "running": "Running upgrade", + "preparing-package": "Preparing package", + "committing-package": "Committing package", + "backing-up": "Creating and verifying backup", + "backed-up": "Backup verified", + "staged": "Package committed", + "migrating": "Starting candidate and checking migration", + "healthy": "Candidate healthy", + "publishing": "Publishing generation", + "completed": "Upgrade completed", + "failed": "Upgrade failed" + }, + "addonFailure": { + "upgrade_failed": "Upgrade failed. Check the owner logs for details.", + "interrupted": "Upgrade interrupted by owner shutdown or restart. It was not replayed.", + "recovery_required": "Upgrade interrupted. Data recovery is required before this add-on can run." + } } } diff --git a/core/http/react-ui/public/locales/fr/operate.json b/core/http/react-ui/public/locales/fr/operate.json index 4b4c3af77..50e7cf0c5 100644 --- a/core/http/react-ui/public/locales/fr/operate.json +++ b/core/http/react-ui/public/locales/fr/operate.json @@ -52,6 +52,26 @@ "rowUpgraded": "Mis à niveau en {{duration}}", "rowUpgradedPlain": "Mis à niveau", "viewInAddons": "Voir dans Modules", - "upgradeWaiting": "En attente du démarrage du service des modules. Prochaine vérification dans {{seconds}} s." + "upgradeWaiting": "En attente du démarrage du service des modules. Prochaine vérification dans {{seconds}} s.", + "addonStage": { + "submitting": "Submitting upgrade", + "accepted": "Accepted by owner", + "running": "Running upgrade", + "preparing-package": "Preparing package", + "committing-package": "Committing package", + "backing-up": "Creating and verifying backup", + "backed-up": "Backup verified", + "staged": "Package committed", + "migrating": "Starting candidate and checking migration", + "healthy": "Candidate healthy", + "publishing": "Publishing generation", + "completed": "Upgrade completed", + "failed": "Upgrade failed" + }, + "addonFailure": { + "upgrade_failed": "Upgrade failed. Check the owner logs for details.", + "interrupted": "Upgrade interrupted by owner shutdown or restart. It was not replayed.", + "recovery_required": "Upgrade interrupted. Data recovery is required before this add-on can run." + } } } diff --git a/core/http/react-ui/public/locales/id/operate.json b/core/http/react-ui/public/locales/id/operate.json index 5df8d85cc..3a8d226d9 100644 --- a/core/http/react-ui/public/locales/id/operate.json +++ b/core/http/react-ui/public/locales/id/operate.json @@ -52,6 +52,26 @@ "rowUpgraded": "Diperbarui dalam {{duration}}", "rowUpgradedPlain": "Diperbarui", "viewInAddons": "Lihat di Add-on", - "upgradeWaiting": "Menunggu layanan add-on dimulai. Pemeriksaan berikutnya dalam {{seconds}} dtk." + "upgradeWaiting": "Menunggu layanan add-on dimulai. Pemeriksaan berikutnya dalam {{seconds}} dtk.", + "addonStage": { + "submitting": "Submitting upgrade", + "accepted": "Accepted by owner", + "running": "Running upgrade", + "preparing-package": "Preparing package", + "committing-package": "Committing package", + "backing-up": "Creating and verifying backup", + "backed-up": "Backup verified", + "staged": "Package committed", + "migrating": "Starting candidate and checking migration", + "healthy": "Candidate healthy", + "publishing": "Publishing generation", + "completed": "Upgrade completed", + "failed": "Upgrade failed" + }, + "addonFailure": { + "upgrade_failed": "Upgrade failed. Check the owner logs for details.", + "interrupted": "Upgrade interrupted by owner shutdown or restart. It was not replayed.", + "recovery_required": "Upgrade interrupted. Data recovery is required before this add-on can run." + } } } diff --git a/core/http/react-ui/public/locales/it/operate.json b/core/http/react-ui/public/locales/it/operate.json index 078b60821..0d4a1ba10 100644 --- a/core/http/react-ui/public/locales/it/operate.json +++ b/core/http/react-ui/public/locales/it/operate.json @@ -52,6 +52,26 @@ "rowUpgraded": "Aggiornato in {{duration}}", "rowUpgradedPlain": "Aggiornato", "viewInAddons": "Vedi in Add-on", - "upgradeWaiting": "In attesa dell'avvio del servizio dei componenti aggiuntivi. Prossimo controllo tra {{seconds}} s." + "upgradeWaiting": "In attesa dell'avvio del servizio dei componenti aggiuntivi. Prossimo controllo tra {{seconds}} s.", + "addonStage": { + "submitting": "Submitting upgrade", + "accepted": "Accepted by owner", + "running": "Running upgrade", + "preparing-package": "Preparing package", + "committing-package": "Committing package", + "backing-up": "Creating and verifying backup", + "backed-up": "Backup verified", + "staged": "Package committed", + "migrating": "Starting candidate and checking migration", + "healthy": "Candidate healthy", + "publishing": "Publishing generation", + "completed": "Upgrade completed", + "failed": "Upgrade failed" + }, + "addonFailure": { + "upgrade_failed": "Upgrade failed. Check the owner logs for details.", + "interrupted": "Upgrade interrupted by owner shutdown or restart. It was not replayed.", + "recovery_required": "Upgrade interrupted. Data recovery is required before this add-on can run." + } } } diff --git a/core/http/react-ui/public/locales/ko/operate.json b/core/http/react-ui/public/locales/ko/operate.json index abe36e529..8514f7a39 100644 --- a/core/http/react-ui/public/locales/ko/operate.json +++ b/core/http/react-ui/public/locales/ko/operate.json @@ -52,6 +52,26 @@ "rowUpgraded": "{{duration}} 만에 업그레이드됨", "rowUpgradedPlain": "업그레이드됨", "viewInAddons": "애드온에서 보기", - "upgradeWaiting": "애드온 서비스가 시작되기를 기다리는 중입니다. {{seconds}}초 후 다시 확인합니다." + "upgradeWaiting": "애드온 서비스가 시작되기를 기다리는 중입니다. {{seconds}}초 후 다시 확인합니다.", + "addonStage": { + "submitting": "Submitting upgrade", + "accepted": "Accepted by owner", + "running": "Running upgrade", + "preparing-package": "Preparing package", + "committing-package": "Committing package", + "backing-up": "Creating and verifying backup", + "backed-up": "Backup verified", + "staged": "Package committed", + "migrating": "Starting candidate and checking migration", + "healthy": "Candidate healthy", + "publishing": "Publishing generation", + "completed": "Upgrade completed", + "failed": "Upgrade failed" + }, + "addonFailure": { + "upgrade_failed": "Upgrade failed. Check the owner logs for details.", + "interrupted": "Upgrade interrupted by owner shutdown or restart. It was not replayed.", + "recovery_required": "Upgrade interrupted. Data recovery is required before this add-on can run." + } } } diff --git a/core/http/react-ui/public/locales/pt-BR/operate.json b/core/http/react-ui/public/locales/pt-BR/operate.json index c850eb208..e04a74dbf 100644 --- a/core/http/react-ui/public/locales/pt-BR/operate.json +++ b/core/http/react-ui/public/locales/pt-BR/operate.json @@ -52,6 +52,26 @@ "rowUpgraded": "Atualizado em {{duration}}", "rowUpgradedPlain": "Atualizado", "viewInAddons": "Ver em Complementos", - "upgradeWaiting": "Aguardando o serviço de complementos iniciar. Próxima verificação em {{seconds}} s." + "upgradeWaiting": "Aguardando o serviço de complementos iniciar. Próxima verificação em {{seconds}} s.", + "addonStage": { + "submitting": "Submitting upgrade", + "accepted": "Accepted by owner", + "running": "Running upgrade", + "preparing-package": "Preparing package", + "committing-package": "Committing package", + "backing-up": "Creating and verifying backup", + "backed-up": "Backup verified", + "staged": "Package committed", + "migrating": "Starting candidate and checking migration", + "healthy": "Candidate healthy", + "publishing": "Publishing generation", + "completed": "Upgrade completed", + "failed": "Upgrade failed" + }, + "addonFailure": { + "upgrade_failed": "Upgrade failed. Check the owner logs for details.", + "interrupted": "Upgrade interrupted by owner shutdown or restart. It was not replayed.", + "recovery_required": "Upgrade interrupted. Data recovery is required before this add-on can run." + } } } diff --git a/core/http/react-ui/public/locales/zh-CN/operate.json b/core/http/react-ui/public/locales/zh-CN/operate.json index 80a6c6566..967daba3c 100644 --- a/core/http/react-ui/public/locales/zh-CN/operate.json +++ b/core/http/react-ui/public/locales/zh-CN/operate.json @@ -52,6 +52,26 @@ "rowUpgraded": "已升级，用时 {{duration}}", "rowUpgradedPlain": "已升级", "viewInAddons": "在插件中查看", - "upgradeWaiting": "正在等待插件服务启动。{{seconds}} 秒后再次检查。" + "upgradeWaiting": "正在等待插件服务启动。{{seconds}} 秒后再次检查。", + "addonStage": { + "submitting": "Submitting upgrade", + "accepted": "Accepted by owner", + "running": "Running upgrade", + "preparing-package": "Preparing package", + "committing-package": "Committing package", + "backing-up": "Creating and verifying backup", + "backed-up": "Backup verified", + "staged": "Package committed", + "migrating": "Starting candidate and checking migration", + "healthy": "Candidate healthy", + "publishing": "Publishing generation", + "completed": "Upgrade completed", + "failed": "Upgrade failed" + }, + "addonFailure": { + "upgrade_failed": "Upgrade failed. Check the owner logs for details.", + "interrupted": "Upgrade interrupted by owner shutdown or restart. It was not replayed.", + "recovery_required": "Upgrade interrupted. Data recovery is required before this add-on can run." + } } } diff --git a/core/http/react-ui/src/components/OperationCard.jsx b/core/http/react-ui/src/components/OperationCard.jsx index ca9f7c179..78c0db543 100644 --- a/core/http/react-ui/src/components/OperationCard.jsx +++ b/core/http/react-ui/src/components/OperationCard.jsx @@ -80,7 +80,7 @@ export default function OperationCard({ operation, onCancel, onPause, onDismiss, const byteLabel = Number.isFinite(operation.currentBytes) && Number.isFinite(operation.totalBytes) && operation.totalBytes > 0 ? `${formatBytes(operation.currentBytes)} / ${formatBytes(operation.totalBytes)}` : '' - const phaseKey = phaseKeys[operation.phase] + const phaseKey = operation.isAddon ? `activity.addonStage.${operation.stage || operation.phase}` : phaseKeys[operation.phase] const etaLabel = formatEta(operation.etaSeconds) const rateLabel = Number.isFinite(operation.bytesPerSecond) && operation.bytesPerSecond > 0 ? `${formatBytes(operation.bytesPerSecond)}/s` @@ -124,9 +124,10 @@ export default function OperationCard({ operation, onCancel, onPause, onDismiss,
{verb} + {operation.isAddon && operation.fromVersion && operation.toVersion && {operation.fromVersion} to {operation.toVersion}} {operation.nodeName && {t('activity.toNode', { node: operation.nodeName })}} {cancelling && {t('activity.cancelling')}} - {failed && {operation.error}} + {failed && {operation.isAddon && operation.failureCode ? t(`activity.addonFailure.${operation.failureCode}`, { defaultValue: operation.error }) : operation.error}} {!failed && phaseKey && {t(phaseKey)}} {/* Phases and byte counters exist only on the managed-artifact path, so a legacy files: gallery model and every backend diff --git a/core/http/react-ui/src/components/addons/actionBook.js b/core/http/react-ui/src/components/addons/actionBook.js index 19c108046..e8eb15769 100644 --- a/core/http/react-ui/src/components/addons/actionBook.js +++ b/core/http/react-ui/src/components/addons/actionBook.js @@ -19,6 +19,10 @@ import { newOperationId } from '../catalog/catalogState.js' export function createActionBook(makeId = newOperationId) { const held = new Map() return { + restore(subject, entry) { + held.set(subject, { ...entry, unknown: true }) + }, + clear() { held.clear() }, begin(subject, key, meta) { const cur = held.get(subject) if (cur && cur.key === key) { diff --git a/core/http/react-ui/src/context/AuthContext.jsx b/core/http/react-ui/src/context/AuthContext.jsx index 2ad8cb737..408ab08b4 100644 --- a/core/http/react-ui/src/context/AuthContext.jsx +++ b/core/http/react-ui/src/context/AuthContext.jsx @@ -1,6 +1,7 @@ import { useContext, useState, useEffect } from 'react' import { apiUrl } from '../utils/basePath' import { AuthContext } from './authState' +import { addonUpgrades } from '../utils/addonUpgrades' export function AuthProvider({ children }) { const [state, setState] = useState({ @@ -35,6 +36,11 @@ export function AuthProvider({ children }) { }, []) const logout = async () => { + // Detach observers and remove stored identities before leaving this page. + // An auth-state effect may never run when navigation unloads the app. + let persistence + try { persistence = window.sessionStorage } catch { /* storage blocked */ } + addonUpgrades.clearSession(persistence) try { await fetch(apiUrl('/api/auth/logout'), { method: 'POST' }) } catch (_) { /* ignore */ } diff --git a/core/http/react-ui/src/contexts/OperationsContext.jsx b/core/http/react-ui/src/contexts/OperationsContext.jsx index d3b9fe668..51756a1ef 100644 --- a/core/http/react-ui/src/contexts/OperationsContext.jsx +++ b/core/http/react-ui/src/contexts/OperationsContext.jsx @@ -1,6 +1,8 @@ import { createContext, useContext, useState, useEffect, useCallback, useMemo, useRef } from 'react' import { operationsApi } from '../utils/api' import { createTransferRateSampler } from '../utils/transferRate' +import { addonUpgrades } from '../utils/addonUpgrades' +import { createAddonsApi } from '../utils/addonsApi' import { useAuth } from '../context/AuthContext' // Serialize ops into a stable comparison key. Each op is a flat map of @@ -29,7 +31,12 @@ export function OperationsProvider({ children, pollInterval = 1000 }) { const [error, setError] = useState(null) const [history, setHistory] = useState([]) const [historyLoading, setHistoryLoading] = useState(false) - const { isAdmin } = useAuth() + const { isAdmin, user, authEnabled, loading: authLoading } = useAuth() + useEffect(() => { + if (authLoading) return + const principal = isAdmin ? (user?.id || (!authEnabled ? "local-owner" : "")) : "" + addonUpgrades.bind(createAddonsApi(), principal, window.sessionStorage) + }, [isAdmin, user?.id, authEnabled, authLoading]) const intervalRef = useRef(null) const lastSerializedRef = useRef('[]') const liveIDsRef = useRef(new Set()) diff --git a/core/http/react-ui/src/utils/addonUpgrades.js b/core/http/react-ui/src/utils/addonUpgrades.js index a6bae8b25..139d643fa 100644 --- a/core/http/react-ui/src/utils/addonUpgrades.js +++ b/core/http/react-ui/src/utils/addonUpgrades.js @@ -7,10 +7,7 @@ // identity per action, never a second submit after an unknown outcome), reads // the operation until it settles, and keeps what it saw so a list can show it. // -// Cancel aborts the request. The owner treats a canceled upgrade request as a -// cancellation (the last good generation is kept), and the store then reads -// the operation once more, so what it reports is the owner's answer and not -// the click. +// Disconnecting only detaches the observer; accepted work belongs to the owner. import { createActionBook, effectOutcome } from '../components/addons/actionBook.js' import { runLifecycle, settleOperation } from '../components/addons/lifecycleFlow.js' import { updateErrorKey, updateView, upgradeBody } from '../components/addons/updateState.js' @@ -18,17 +15,20 @@ import { AddonsError } from './addonsApi.js' // How far along an operation phase is, for the progress bar. The owner reports // phases, not percentages. -const PHASE_PROGRESS = { intent: 10, prepared: 50, applied: 85 } +const PHASE_PROGRESS = {} const MAX_RECORDS = 50 const POLL_MS = 2000 const POLL_ATTEMPTS = 300 export function phaseProgress(phase) { - return PHASE_PROGRESS[phase] || 5 + return PHASE_PROGRESS[phase] || 0 } export function createUpgradeStore({ now = Date.now, waits, sleep, book = createActionBook() } = {}) { let snapshot = { live: [], records: [] } + let scope = "" + let storage = null + let epoch = 0 const listeners = new Set() const controllers = new Map() const cancelled = new Set() @@ -51,6 +51,7 @@ export function createUpgradeStore({ now = Date.now, waits, sleep, book = create const emit = (next) => { snapshot = next + if (storage && scope) { try { storage.setItem(`addon-upgrades:${scope}`, JSON.stringify(next.live.map(({ addonId, operationId, jobID, fromVersion, toVersion, name, startedAt, baseGeneration }) => ({ addonId, operationId, jobID, fromVersion, toVersion, name, startedAt, baseGeneration })))) } catch { /* unavailable storage only loses observer recovery */ } } for (const listener of listeners) listener() } const patch = (jobID, fields) => { @@ -64,12 +65,48 @@ export function createUpgradeStore({ now = Date.now, waits, sleep, book = create ...api, async operationStatus(...args) { const status = await api.operationStatus(...args) - if (status && status.phase) patch(jobID, { phase: status.phase, progress: phaseProgress(status.phase) }) + if (status && status.phase) patch(jobID, { phase: status.phase, stage: status.stage, failureCode: status.failureCode, message: status.stage || status.phase, progress: 0 }) return status }, }) const store = { + bind(api, principal, persistence) { + if (scope === principal && storage === persistence) return + const previous = scope + epoch += 1 + for (const controller of controllers.values()) controller.abort() + controllers.clear() + book.clear() + if (!principal && storage && previous) { try { storage.removeItem(`addon-upgrades:${previous}`) } catch { /* storage unavailable */ } } + scope = principal + storage = persistence + snapshot = { live: [], records: [] } + for (const listener of listeners) listener() + if (!scope || !storage) return + let pending = [] + try { pending = JSON.parse(storage.getItem(`addon-upgrades:${scope}`) || "[]") } catch { return } + if (!Array.isArray(pending)) return + for (const op of pending.slice(0, 20)) { + if (!op.operationId || !op.addonId) continue + book.restore(op.addonId, { key: `upgrade:${op.addonId}:${op.baseGeneration}`, operationId: op.operationId }) + const controller = new AbortController() + controllers.set(op.jobID, controller) + emit({ ...snapshot, live: [...snapshot.live, { ...op, isAddon: true, taskType: "upgrade", state: "running", progress: 0, cancellable: false }] }) + void store.follow(api, { jobID: op.jobID, addon: { id: op.addonId, generation: op.baseGeneration }, operationId: op.operationId, key: `upgrade:${op.addonId}:${op.baseGeneration}`, controller, submit: false }) + } + }, + + // Logout must not rely on a React state update before navigation, and may + // run before this store has bound a principal. Clear only our namespace. + clearSession(persistence) { + store.bind(null, '', null) + try { + const keys = Array.from({ length: persistence?.length || 0 }, (_, i) => persistence.key(i)) + for (const key of keys) if (key?.startsWith('addon-upgrades:')) persistence.removeItem(key) + } catch { /* blocked storage must not prevent logout */ } + }, + subscribe(listener) { listeners.add(listener) return () => listeners.delete(listener) @@ -94,6 +131,7 @@ export function createUpgradeStore({ now = Date.now, waits, sleep, book = create fullName: addon.id, isAddon: true, taskType: 'upgrade', + baseGeneration: addon.generation, fromVersion: addon.version || '', toVersion: view.version, startedAt: new Date(now()).toISOString(), @@ -110,7 +148,7 @@ export function createUpgradeStore({ now = Date.now, waits, sleep, book = create controllers.set(jobID, controller) emit({ ...snapshot, - live: [...withoutOld, { ...base, jobID, operationId, state: 'running', phase: 'intent', progress: phaseProgress('intent'), cancellable: true, error: '', errorKey: '' }], + live: [...withoutOld, { ...base, jobID, operationId, state: 'running', phase: 'submitting', progress: 0, cancellable: false, error: '', errorKey: '' }], }) return store.follow(api, { jobID, addon, operationId, key, controller, submit: true }) }, @@ -118,6 +156,7 @@ export function createUpgradeStore({ now = Date.now, waits, sleep, book = create // Runs the submit (or, with submit false, only reads the earlier operation) // and turns its end into the entry of a list. async follow(api, { jobID, addon, operationId, key, controller, submit }) { + const followingEpoch = epoch const signal = controller.signal const tracked = trackPhases(api, jobID) try { @@ -151,8 +190,10 @@ export function createUpgradeStore({ now = Date.now, waits, sleep, book = create verified = undefined } } + if (followingEpoch !== epoch) return { ok: false, outcome: "detached", message: "" } return store.conclude(jobID, addon, key, effectOutcome(outcome, verified), operationId) } catch (e) { + if (followingEpoch !== epoch) return { ok: false, outcome: "detached", message: "" } if (e && e.name === 'AbortError' && cancelled.has(jobID)) return store.afterCancel(api, { jobID, addon, operationId, key }) const kind = e && e.kind ? e.kind : 'unavailable' if (kind === 'denied' || kind === 'invalid' || kind === 'auth') book.resolve(addon.id, key) @@ -216,7 +257,7 @@ export function createUpgradeStore({ now = Date.now, waits, sleep, book = create cancel(jobID) { const controller = controllers.get(jobID) - if (!controller) return false + if (!controller || !snapshot.live.find(op => op.jobID === jobID)?.cancellable) return false cancelled.add(jobID) patch(jobID, { state: 'cancelling', cancellable: false }) controller.abort() diff --git a/core/http/react-ui/src/utils/addonsApi.js b/core/http/react-ui/src/utils/addonsApi.js index 9ea0778d7..91e3c3748 100644 --- a/core/http/react-ui/src/utils/addonsApi.js +++ b/core/http/react-ui/src/utils/addonsApi.js @@ -30,6 +30,8 @@ export function normalizeOperation(payload) { operationId: p.OperationID || p.operationId || '', addonId: p.AddonID || p.addonId || '', phase: p.Phase || p.phase || '', + stage: p.stage || '', + failureCode: p.failureCode || '', generation: Number(p.Generation ?? p.generation ?? 0), runtimeState: p.runtimeState || '', } diff --git a/core/http/react-ui/tests/unit/updatereview.test.mjs b/core/http/react-ui/tests/unit/updatereview.test.mjs index abe15784c..f1aa1be50 100644 --- a/core/http/react-ui/tests/unit/updatereview.test.mjs +++ b/core/http/react-ui/tests/unit/updatereview.test.mjs @@ -92,26 +92,19 @@ test('upgrade store: an add-on without a confirmed candidate, or one already run void first }) -test('upgrade store: cancel aborts the request and reports what the owner says, not the click', async () => { - const abortable = (signal) => new Promise((_, reject) => signal.addEventListener('abort', () => reject(new DOMException('aborted', 'AbortError')))) - const owner = fakeOwner({ fate: { notes: abortable, cal: abortable }, status: async () => ({ phase: 'failed' }) }) +test('upgrade store: owner-managed upgrades report stages, not synthetic percentages or cancellation', async () => { + let finish + const owner = fakeOwner({ fate: { notes: () => new Promise(resolve => { finish = resolve }) } }) const s = store() const run = s.start(owner.api, pendingAddon('notes')) await new Promise(r => setImmediate(r)) const job = s.getSnapshot().live[0] - assert.equal(job.cancellable, true) - assert.equal(s.cancel(job.jobID), true) - assert.equal(s.getSnapshot().live[0].state, 'cancelling') - const out = await run - assert.equal(out.outcome, 'cancelled') - assert.equal(s.getSnapshot().live.length, 0) - assert.equal(s.getSnapshot().records[0].outcome, 'cancelled') - // The owner finished it before the cancel landed: that is an upgrade. - const owner2 = fakeOwner({ fate: { cal: abortable }, status: async () => ({ phase: 'completed' }) }) - const run2 = s.start(owner2.api, pendingAddon('cal')) - await new Promise(r => setImmediate(r)) - s.cancel(s.getSnapshot().live[0].jobID) - assert.equal((await run2).ok, true) + assert.equal(job.cancellable, false) + assert.equal(job.progress, 0) + assert.equal(job.phase, 'submitting') + assert.equal(s.cancel(job.jobID), false) + finish('applied') + assert.equal((await run).ok, true) }) test('upgrade store: an unknown outcome is read on retry and never submitted a second time', async () => { @@ -201,3 +194,57 @@ test('upgrade store: check now ends the wait of a followed upgrade', async () => assert.equal(out.ok, true) assert.equal(s.checkNow(op.jobID), false) }) + + +test('upgrade observers resume the same ID after reload and isolate account changes/logout', async () => { + const values = new Map() + const storage = { getItem: k => values.get(k), setItem: (k,v) => values.set(k,v), removeItem: k => values.delete(k) } + const api = { updates: { upgrade: async () => ({ phase: 'accepted' }) }, operationStatus: async () => ({ phase: 'running', stage: 'backing-up' }), installed: async () => [] } + const first = store() + first.bind(api, 'alice', storage) + await first.start(api, pendingAddon('notes')) + const id = first.getSnapshot().live[0].operationId + let reads = [] + const resumedApi = { ...api, operationStatus: async op => { reads.push(op); return { phase: 'failed', failureCode: 'interrupted' } } } + const reloaded = store() + reloaded.bind(resumedApi, 'bob', storage) + assert.equal(reloaded.getSnapshot().live.length, 0) + reloaded.bind(resumedApi, 'alice', storage) + await new Promise(r => setImmediate(r)) + assert.deepEqual(reads, [id]) + assert.equal(reloaded.getSnapshot().live[0].failureCode, 'interrupted') + reloaded.bind(resumedApi, '', storage) + assert.equal(reloaded.getSnapshot().live.length, 0) + assert.equal(values.has('addon-upgrades:alice'), false) +}) + + +test('account change gives Bob a fresh action identity and restores Alice unknown identity on return', async () => { + const values = new Map() + const storage = { getItem: k => values.get(k), setItem: (k,v) => values.set(k,v), removeItem: k => values.delete(k) } + const owner = fakeOwner({ fate: { notes: async () => { throw new AddonsError({ kind: 'unknown_outcome' }) } }, status: async () => ({ phase: 'running' }) }) + const s = store() + s.bind(owner.api, 'alice', storage) + await s.start(owner.api, pendingAddon('notes')) + const aliceID = owner.calls[0].body.operationId + s.bind(owner.api, 'bob', storage) + await s.start(owner.api, pendingAddon('notes')) + const bobID = owner.calls[1].body.operationId + assert.notEqual(bobID, aliceID) + s.bind(owner.api, 'alice', storage) + await new Promise(r => setImmediate(r)) + const job = s.getSnapshot().live[0] + assert.equal(job.operationId, aliceID) + await s.retry(owner.api, job.jobID) + assert.equal(owner.calls.length, 2, 'restored unknown operation must only poll, never resubmit') +}) + +test('logout cleanup clears persisted observers even in a fresh unbound store before navigation', () => { + const values = new Map([['addon-upgrades:alice', '[{"operationId":"private"}]'], ['unrelated', 'keep']]) + const storage = { get length() { return values.size }, key: i => [...values.keys()][i], getItem: k => values.get(k), setItem: (k,v) => values.set(k,v), removeItem: k => values.delete(k) } + const fresh = store() + fresh.clearSession(storage) + assert.equal(values.has('addon-upgrades:alice'), false) + assert.equal(values.get('unrelated'), 'keep') + assert.deepEqual(fresh.getSnapshot(), { live: [], records: [] }) +}) diff --git a/core/http/react-ui/tests/unit/updates.test.mjs b/core/http/react-ui/tests/unit/updates.test.mjs index a346c5804..6f88b8cc3 100644 --- a/core/http/react-ui/tests/unit/updates.test.mjs +++ b/core/http/react-ui/tests/unit/updates.test.mjs @@ -86,7 +86,7 @@ test('updates: the client sends check and upgrade as the administrator\'s own ac assert.equal(calls[1].init.headers['X-LocalAI-Confirm'], 'chk-1') assert.equal(calls[1].init.credentials, 'same-origin') const done = await api.updates.upgrade('notes', { operationId: 'up-9', expectedDigest: 'sha256:a', candidateDigest: 'sha256:b', generation: 1 }) - assert.deepEqual(done, { operationId: 'op-9', addonId: 'notes', phase: 'completed', generation: 2, runtimeState: '' }) + assert.deepEqual(done, { operationId: 'op-9', addonId: 'notes', phase: 'completed', generation: 2, runtimeState: '', stage: '', failureCode: '' }) assert.equal(calls[2].init.headers['X-LocalAI-Confirm'], 'up-9') assert.deepEqual(JSON.parse(calls[2].init.body), { operationId: 'up-9', expectedDigest: 'sha256:a', candidateDigest: 'sha256:b', generation: 1 }) await assert.rejects(api.updates.upgrade('notes', { operationId: 'stale-1', expectedDigest: 'stale', candidateDigest: 'x' }), (e) => e instanceof AddonsError && e.kind === 'invalid' && e.type === 'stale_installed' && e.status === 409) diff --git a/core/services/addons/gateway/gateway.go b/core/services/addons/gateway/gateway.go index 9b3938350..2b27c509d 100644 --- a/core/services/addons/gateway/gateway.go +++ b/core/services/addons/gateway/gateway.go @@ -144,7 +144,13 @@ func (g *Gateway) dispatchLocal(ctx context.Context, actor contract.Actor, reque // disable; a stale one is denied before anything is recorded. cmd.Generation = request.Generation } - status, err := g.Manager.Submit(ctx, actor, cmd) + var status addons.OperationStatus + var err error + if cmd.Action == "upgrade" { + status, err = g.Manager.SubmitUpgrade(ctx, actor, cmd) + } else { + status, err = g.Manager.Submit(ctx, actor, cmd) + } if err != nil { return contract.Result{}, err } diff --git a/core/services/addons/gateway/updates.go b/core/services/addons/gateway/updates.go index 9bdc6226e..0519a9954 100644 --- a/core/services/addons/gateway/updates.go +++ b/core/services/addons/gateway/updates.go @@ -107,7 +107,7 @@ func (g *Gateway) updateAction(ctx context.Context, actor contract.Actor, reques err = addons.ErrConflict break } - data, err = g.Manager.Submit(ctx, actor, cmd) + data, err = g.Manager.SubmitUpgrade(ctx, actor, cmd) default: return contract.Result{}, ErrForbidden } diff --git a/core/services/addons/gateway/upgrade_polling_test.go b/core/services/addons/gateway/upgrade_polling_test.go new file mode 100644 index 000000000..6e0ce647f --- /dev/null +++ b/core/services/addons/gateway/upgrade_polling_test.go @@ -0,0 +1,97 @@ +// SPDX-License-Identifier: MIT +package gateway + +import ( + "context" + "encoding/json" + "net/http" + "os" + "path/filepath" + "time" + + "github.com/mudler/LocalAI/core/services/addons" + "github.com/mudler/LocalAI/core/services/addons/recovery" + "github.com/mudler/LocalAI/pkg/addons/contract" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +type blockedUpgradePolicy struct{ entered, release chan struct{} } + +func (*blockedUpgradePolicy) Mode() string { return "named-user" } +func (*blockedUpgradePolicy) Hold(context.Context, string) (func(), error) { return func() {}, nil } +func (p *blockedUpgradePolicy) Export(context.Context, string) (recovery.PolicyExport, error) { + close(p.entered) + <-p.release + return recovery.PolicyExport{}, nil +} +func (*blockedUpgradePolicy) Restore(context.Context, string, recovery.PolicyExport, recovery.RestoreOptions) (recovery.RestoreReport, error) { + return recovery.RestoreReport{}, nil +} +func (*blockedUpgradePolicy) LostWrites(context.Context, string, time.Time) (int, error) { + return 0, nil +} +func (*blockedUpgradePolicy) Forget(context.Context, string) (int, error) { return 0, nil } + +var _ = Describe("durable upgrade carrier polling", func() { + It("accepts before a slow backup, detaches the request and polls through either frontend", func() { + actor := contract.Actor{UserID: "u", CredentialID: "k", Mode: "named-user"} + manager := ownerManager(actor, true) + st, e := manager.Status(context.Background(), "sample") + Expect(e).NotTo(HaveOccurred()) + Expect(os.MkdirAll(st.DataDirectory, 0700)).To(Succeed()) + Expect(os.WriteFile(filepath.Join(st.DataDirectory, "data"), []byte("retained"), 0600)).To(Succeed()) + policy := &blockedUpgradePolicy{make(chan struct{}), make(chan struct{})} + defer func() { + select { + case <-policy.release: + default: + close(policy.release) + } + }() + r, e := recovery.New(recovery.Config{Root: filepath.Dir(filepath.Dir(st.DataDirectory)), Host: manager.RecoveryHost(), Policy: policy}) + Expect(e).NotTo(HaveOccurred()) + manager.SetRecovery(r) + _, cs := carrierFixture(func(c *Carrier) http.Handler { + h, e := c.OwnerHandler(&Gateway{Manager: manager, Revalidate: func(context.Context, contract.Actor) error { return nil }}) + Expect(e).NotTo(HaveOccurred()) + return h + }) + // Bound the old ordinary client deadline, rather than sleeping 65 seconds. + cs["front-a"].client.Timeout = 100 * time.Millisecond + request, cancel := context.WithCancel(context.Background()) + cmd := contract.Request{AddonID: "sample", OperationID: "upgrade", Action: "upgrade", Arguments: []byte(`{"sourceId":"test","version":"1.2.3"}`)} + out, e := cs["front-a"].Dispatch(request, actor, cmd) + Expect(e).NotTo(HaveOccurred()) + cancel() + var accepted addons.OperationStatus + Expect(json.Unmarshal(out.Payload, &accepted)).To(Succeed()) + Expect(accepted.Phase).To(Equal("accepted")) + Eventually(policy.entered).Should(BeClosed()) + // Work remains blocked beyond the old deadline while status reads succeed. + Consistently(func() string { + out, e := cs["front-b"].Dispatch(context.Background(), actor, contract.Request{AddonID: "sample", OperationID: "upgrade", Action: "operation-status", Arguments: []byte(`{}`)}) + Expect(e).NotTo(HaveOccurred()) + var status addons.OperationStatus + Expect(json.Unmarshal(out.Payload, &status)).To(Succeed()) + return status.Stage + }, 150*time.Millisecond, 20*time.Millisecond).Should(Equal("backing-up")) + _, e = cs["front-b"].Dispatch(context.Background(), actor, cmd) + Expect(e).NotTo(HaveOccurred()) + changed := cmd + changed.Arguments = []byte(`{"sourceId":"test","version":"9.0.0"}`) + _, e = cs["front-b"].Dispatch(context.Background(), actor, changed) + Expect(e).To(HaveOccurred()) + close(policy.release) + Eventually(func() string { + out, e := cs["front-a"].Dispatch(context.Background(), actor, contract.Request{AddonID: "sample", OperationID: "upgrade", Action: "operation-status", Arguments: []byte(`{}`)}) + Expect(e).NotTo(HaveOccurred()) + var status addons.OperationStatus + Expect(json.Unmarshal(out.Payload, &status)).To(Succeed()) + return status.Phase + }).Should(Equal("completed")) + st, e = manager.Status(context.Background(), "sample") + Expect(e).NotTo(HaveOccurred()) + Expect(st.Generation).To(Equal(uint64(2))) + }) +}) diff --git a/core/services/addons/journal.go b/core/services/addons/journal.go index 728e041f4..5541f15e8 100644 --- a/core/services/addons/journal.go +++ b/core/services/addons/journal.go @@ -10,6 +10,7 @@ import ( ) type journalOperation struct { + Async bool Command LifecycleCommand Actor contract.Actor Status OperationStatus @@ -50,6 +51,11 @@ func durableWrite(path string, b []byte) error { return d.Sync() } func (m *Manager) persist(phase string) error { + err := m.persistJournal(phase) + m.refreshUpgradeViews() + return err +} +func (m *Manager) persistJournal(phase string) error { b, e := json.Marshal(m.journal) if e == nil { e = durableWrite(filepath.Join(m.config.Root, "journal.json"), b) diff --git a/core/services/addons/manager.go b/core/services/addons/manager.go index 5a2ae1c80..84afe0585 100644 --- a/core/services/addons/manager.go +++ b/core/services/addons/manager.go @@ -99,7 +99,13 @@ type Migrator interface { type Manager struct { // One lock also serializes journal publication across addons. No operation can // overtake another operation on the same addon, including reconciliation. - mu sync.Mutex + mu sync.Mutex + upgradeMu sync.Mutex + upgradeViews map[string]upgradeSnapshot + upgradeCtx context.Context + upgradeCancel context.CancelFunc + upgradeWG sync.WaitGroup + upgradeActive, upgradeClosing bool // opMu serializes Submit and Reconcile as whole operations, so mu can be // released around a slow fetch without letting another operation in. opMu sync.Mutex @@ -228,7 +234,8 @@ func NewManager(ctx context.Context, cfg OwnerConfig, c catalog.Catalog, i *pack if e := cfg.validate(); e != nil { return nil, e } - m := &Manager{config: cfg, catalog: c, installer: i, admin: admin} + m := &Manager{config: cfg, catalog: c, installer: i, admin: admin, upgradeViews: map[string]upgradeSnapshot{}} + m.upgradeCtx, m.upgradeCancel = context.WithCancel(context.Background()) if cfg.InstanceID != cfg.OwnerID { return m, nil } @@ -244,9 +251,15 @@ func NewManager(ctx context.Context, cfg OwnerConfig, c catalog.Catalog, i *pack _ = release() return nil, e } + m.refreshUpgradeViews() return m, nil } func (m *Manager) Close() error { + m.upgradeMu.Lock() + m.upgradeClosing = true + m.upgradeCancel() + m.upgradeMu.Unlock() + m.upgradeWG.Wait() m.mu.Lock() defer m.mu.Unlock() if m.closed && m.release == nil { @@ -306,7 +319,7 @@ func (m *Manager) Status(ctx context.Context, id string) (Status, error) { s.ObservedHealth = "stopped" s.StartEligible = s.Installed && s.DesiredEnabled for _, op := range m.journal.Operations { - if op.Command.AddonID == id && op.Status.Phase != "completed" && op.Status.Phase != "failed" { + if op.Command.AddonID == id && op.Status.Phase != string(lifecycleCompleted) && op.Status.Phase != string(lifecycleFailed) { s.StartEligible = false } } @@ -349,6 +362,7 @@ func (m *Manager) submit(ctx context.Context, actor contract.Actor, cmd Lifecycl } var op journalOperation resumed := false + accepted := false if !addonID.MatchString(cmd.AddonID) || cmd.OperationID == "" || len(cmd.OperationID) > 256 { return OperationStatus{}, ErrConflict } @@ -356,7 +370,13 @@ func (m *Manager) submit(ctx context.Context, actor contract.Actor, cmd Lifecycl if old.Command != cmd || old.Actor != actor { return OperationStatus{}, ErrConflict } - if old.Status.Phase == "failed" { + if old.Async && old.Status.Phase == string(lifecycleAccepted) { + if ctx.Value(upgradeExecutionKey{}) != m { + return old.Status, ErrRecoveryRequired + } + accepted = true + op = old + } else if old.Status.Phase == string(lifecycleFailed) { // The answer stays the failure, but the add-on must not stay // down because the failing attempt could not put it back. if e := m.syncRuntime(ctx, cmd.AddonID); e != nil { @@ -364,11 +384,13 @@ func (m *Manager) submit(ctx context.Context, actor contract.Actor, cmd Lifecycl } return old.Status, ErrOperationFailed } - if _, again := m.canceled[cmd.OperationID]; again && old.Status.Phase == "intent" { + if accepted { + // Accepted work continues exactly once under the owner worker. + } else if _, again := m.canceled[cmd.OperationID]; again && old.Status.Phase == string(lifecycleIntent) { resumed = true op = old } else { - if old.Status.Phase != "completed" { + if old.Status.Phase != string(lifecycleCompleted) { return old.Status, ErrRecoveryRequired } return old.Status, m.syncFor(ctx, cmd.AddonID) @@ -378,20 +400,20 @@ func (m *Manager) submit(ctx context.Context, actor contract.Actor, cmd Lifecycl if pinned && (cmd.Action != "upgrade" || cmd.ExpectedDigest == "" || cmd.CandidateDigest == "") { return OperationStatus{}, ErrConflict } - if pinned && !resumed { + if pinned && !resumed && !accepted { // The same transition under another operation ID is a duplicate, // whether the first one is in flight or done. Only a failed attempt // leaves the transition open for a new one. for id, other := range m.journal.Operations { c := other.Command - if id != cmd.OperationID && c.Action == "upgrade" && c.AddonID == cmd.AddonID && c.ExpectedDigest == cmd.ExpectedDigest && c.CandidateDigest == cmd.CandidateDigest && other.Status.Phase != "failed" { + if id != cmd.OperationID && c.Action == "upgrade" && c.AddonID == cmd.AddonID && c.ExpectedDigest == cmd.ExpectedDigest && c.CandidateDigest == cmd.CandidateDigest && other.Status.Phase != string(lifecycleFailed) { return OperationStatus{}, ErrTransitionConflict } } } - if !resumed { + if !resumed && !accepted { for _, other := range m.journal.Operations { - if other.Command.AddonID == cmd.AddonID && other.Status.Phase != "completed" && other.Status.Phase != "failed" { + if other.Command.AddonID == cmd.AddonID && other.Status.Phase != string(lifecycleCompleted) && other.Status.Phase != string(lifecycleFailed) { return OperationStatus{}, ErrRecoveryRequired } } @@ -429,10 +451,17 @@ func (m *Manager) submit(ctx context.Context, actor contract.Actor, cmd Lifecycl default: return OperationStatus{}, ErrConflict } - if !resumed { - op = journalOperation{Command: cmd, Actor: actor, Status: OperationStatus{OperationID: cmd.OperationID, AddonID: cmd.AddonID, Phase: "intent"}} + if accepted { + phase, _ := upgradeTransition(lifecyclePhase(op.Status.Phase), upgradeStart) + op.Status.Phase = string(phase) m.journal.Operations[cmd.OperationID] = op - if e := m.persist("intent"); e != nil { + if e := m.persist(string(lifecycleRunning)); e != nil { + return op.Status, e + } + } else if !resumed { + op = journalOperation{Command: cmd, Actor: actor, Status: OperationStatus{OperationID: cmd.OperationID, AddonID: cmd.AddonID, Phase: string(lifecycleIntent)}} + m.journal.Operations[cmd.OperationID] = op + if e := m.persist(string(lifecycleIntent)); e != nil { return op.Status, e } } @@ -443,9 +472,19 @@ func (m *Manager) submit(ctx context.Context, actor contract.Actor, cmd Lifecycl // The client gets a typed cause or the generic error; the operator // gets the underlying error here, which would otherwise be lost. xlog.Warn("Addon operation failed", "action", cmd.Action, "addonId", cmd.AddonID, "operationId", cmd.OperationID, "causes", describeCauses(cause)) - op.Status.Phase = "failed" + op.Status.Stage = m.journal.Operations[cmd.OperationID].Status.Stage + op.Status.Phase = string(lifecycleFailed) + op.Status.FailureCode = "upgrade_failed" + for _, c := range cause { + if errors.Is(c, context.Canceled) || errors.Is(c, context.DeadlineExceeded) { + op.Status.FailureCode = "interrupted" + } + if errors.Is(c, recovery.ErrNeedsRecovery) { + op.Status.FailureCode = "recovery_required" + } + } m.journal.Operations[cmd.OperationID] = op - if e := m.persist("failed"); e != nil { + if e := m.persist(string(lifecycleFailed)); e != nil { return op.Status, e } for _, c := range cause { @@ -482,6 +521,9 @@ func (m *Manager) submit(ctx context.Context, actor contract.Actor, cmd Lifecycl // available, and opMu keeps every other operation out. var r catalog.Release var resolveErr error + if e := m.upgradeStage(cmd.OperationID, "preparing-package"); e != nil { + return op.Status, e + } m.mu.Unlock() if release != nil { r = *release @@ -530,6 +572,9 @@ func (m *Manager) submit(ctx context.Context, actor contract.Actor, cmd Lifecycl return m.upgradeWithRecovery(ctx, cmd, op, &next, staged, fail) } } + if e = m.upgradeStage(cmd.OperationID, "committing-package"); e != nil { + return op.Status, e + } receipt, e = m.installer.Commit(ctx, staged) if e != nil { return fail(&FailureCause{Code: CauseInstallStorage, detail: fmt.Errorf("commit: %w", e)}) @@ -584,9 +629,9 @@ func (m *Manager) submit(ctx context.Context, actor contract.Actor, cmd Lifecycl } op.Next = &next op.Status.Generation = next.Generation - op.Status.Phase = "prepared" + op.Status.Phase = string(lifecyclePrepared) m.journal.Operations[cmd.OperationID] = op - if e := m.persist("prepared"); e != nil { + if e := m.persist(string(lifecyclePrepared)); e != nil { return op.Status, e } result, e := m.apply(op) @@ -650,15 +695,15 @@ func (m *Manager) resolveRelease(ctx context.Context, cmd LifecycleCommand) (cat func (m *Manager) apply(op journalOperation) (OperationStatus, error) { m.journal.Addons[op.Command.AddonID] = copyStatus(*op.Next) - op.Status.Phase = "applied" + op.Status.Phase = string(lifecycleApplied) m.journal.Operations[op.Command.OperationID] = op - if e := m.persist("applied"); e != nil { + if e := m.persist(string(lifecycleApplied)); e != nil { return op.Status, e } - op.Status.Phase = "completed" + op.Status.Phase = string(lifecycleCompleted) op.Next = nil m.journal.Operations[op.Command.OperationID] = op - return op.Status, m.persist("completed") + return op.Status, m.persist(string(lifecycleCompleted)) } func (m *Manager) Reconcile(ctx context.Context) error { m.opMu.Lock() @@ -684,26 +729,36 @@ func (m *Manager) Reconcile(ctx context.Context) error { if e := ctx.Err(); e != nil { return e } - switch op.Status.Phase { - case "intent": - op.Status.Phase = "failed" + switch lifecyclePhase(op.Status.Phase) { + case lifecycleAccepted, lifecycleRunning, lifecycleIntent: + op.Status.Phase = string(lifecycleFailed) + op.Status.FailureCode = "interrupted" m.journal.Operations[id] = op - if e := m.persist("failed"); e != nil { + if e := m.persist(string(lifecycleFailed)); e != nil { return e } - case "prepared", "applied": + case lifecyclePrepared, lifecycleApplied: if op.Next == nil { return ErrConfiguration } if _, e := m.apply(op); e != nil { return e } - case "completed", "failed": + case lifecycleCompleted, lifecycleFailed: default: return ErrConfiguration } } m.reconcileData(ctx) + for id, op := range m.journal.Operations { + if op.Async && op.Status.Phase == string(lifecycleFailed) && m.dataBlocked(op.Command.AddonID) { + op.Status.FailureCode = "recovery_required" + m.journal.Operations[id] = op + } + } + if e := m.persist("reconciled"); e != nil { + return e + } return m.syncRuntime(ctx, "") } func syncPackage(root string) error { @@ -1050,6 +1105,25 @@ func (m *Manager) Connector(ctx context.Context, addonID string, req contract.Co // OperationStatus only observes lifecycle intent. It never resumes a command or // infers runtime completion from a completed journal publication. func (m *Manager) OperationStatus(ctx context.Context, actor contract.Actor, id string) (OperationStatus, error) { + m.upgradeMu.Lock() + view, found := m.upgradeViews[id] + closing := m.upgradeClosing + m.upgradeMu.Unlock() + if found { + if closing { + return OperationStatus{}, ErrNotOwner + } + if m.admin == nil || m.admin(ctx, actor) != nil { + return OperationStatus{}, ErrForbidden + } + if view.Actor != actor { + return OperationStatus{}, ErrOperationNotFound + } + if err := ctx.Err(); err != nil { + return OperationStatus{}, err + } + return view.Status, nil + } // A slow lifecycle publication must not strand disconnected status readers. for !m.mu.TryLock() { timer := time.NewTimer(10 * time.Millisecond) @@ -1073,7 +1147,7 @@ func (m *Manager) OperationStatus(ctx context.Context, actor contract.Actor, id } result := OperationStatus{OperationID: id, AddonID: op.Status.AddonID, Phase: op.Status.Phase, Generation: op.Status.Generation, RuntimeState: "not-observed"} if m.uncertain || m.runtimeRecoveryRequired { - result.RuntimeState = "recovery-required" + result.RuntimeState = string(lifecycleRecovery) } return result, nil } diff --git a/core/services/addons/recovery.go b/core/services/addons/recovery.go index d500ec352..305fb03f5 100644 --- a/core/services/addons/recovery.go +++ b/core/services/addons/recovery.go @@ -218,9 +218,9 @@ func (m *Manager) upgradeWithRecovery(ctx context.Context, cmd LifecycleCommand, } op.Next = next op.Status.Generation = next.Generation - op.Status.Phase = "prepared" + op.Status.Phase = string(lifecyclePrepared) m.journal.Operations[cmd.OperationID] = op - if e := m.persist("prepared"); e != nil { + if e := m.persist(string(lifecyclePrepared)); e != nil { applyErr = e result = op.Status return e @@ -234,7 +234,7 @@ func (m *Manager) upgradeWithRecovery(ctx context.Context, cmd LifecycleCommand, applied = true return nil } - _, e := m.recovery.With(host).UpgradeReceipt(ctx, cmd.AddonID, staged) + _, e := m.recovery.With(host).WithProgress(func(stage string) error { return m.upgradeStage(cmd.OperationID, stage) }).UpgradeReceipt(ctx, cmd.AddonID, staged) switch { case applyErr != nil: if m.runtime != nil { @@ -246,7 +246,7 @@ func (m *Manager) upgradeWithRecovery(ctx context.Context, cmd LifecycleCommand, m.syncSkills(ctx, cmd.AddonID) return result, m.syncFor(ctx, cmd.AddonID) case e != nil: - return fail() + return fail(e) } return fail() } diff --git a/core/services/addons/recovery/backup.go b/core/services/addons/recovery/backup.go index d32a8a528..063c54c32 100644 --- a/core/services/addons/recovery/backup.go +++ b/core/services/addons/recovery/backup.go @@ -85,8 +85,9 @@ type BackupReceipt struct { // Recovery owns the backup, upgrade, restore and purge of add-on data. type Recovery struct { - cfg Config - locks *lockSet + progress func(string) error + cfg Config + locks *lockSet // Test-only seams; nil in production. free func(path string) (uint64, error) fail func(phase string) error @@ -123,6 +124,13 @@ func (r *Recovery) With(host Host) *Recovery { return &c } +// WithProgress observes durable upgrade stages without changing recovery rules. +func (r *Recovery) WithProgress(f func(string) error) *Recovery { + c := *r + c.progress = f + return &c +} + // SetPolicy sets the policy port once the owner has one. func (r *Recovery) SetPolicy(p PolicyPort) { r.cfg.Policy = p } diff --git a/core/services/addons/recovery/migration.go b/core/services/addons/recovery/migration.go index ee3aa8c26..55a8f0e75 100644 --- a/core/services/addons/recovery/migration.go +++ b/core/services/addons/recovery/migration.go @@ -137,7 +137,13 @@ func (r *Recovery) UpgradeReceipt(ctx context.Context, addonID string, staged Ca } phase := func(p string) error { j.Phase = p - return r.saveJournal(addonID, j) + if e := r.saveJournal(addonID, j); e != nil { + return e + } + if r.progress != nil { + return r.progress(p) + } + return nil } var q Quiesced quiesced := false @@ -153,6 +159,11 @@ func (r *Recovery) UpgradeReceipt(ctx context.Context, addonID string, staged Ca defer stop() err = r.undoUpgrade(dctx, host, addonID, info, j, q, quiesced && !published, err) }() + if r.progress != nil { + if err = r.progress("backing-up"); err != nil { + return packages.PackageReceipt{}, err + } + } if err = r.hook("before-backup"); err != nil { return packages.PackageReceipt{}, err } diff --git a/core/services/addons/state.go b/core/services/addons/state.go index 08a9c6ed1..b4587e2d1 100644 --- a/core/services/addons/state.go +++ b/core/services/addons/state.go @@ -3,6 +3,21 @@ package addons import "github.com/mudler/LocalAI/core/services/addons/packages" +// lifecyclePhase is shared by the owner journal and upgrade reducer. The +// operation DTO retains string fields for compatibility with existing callers. +type lifecyclePhase string + +const ( + lifecycleAccepted lifecyclePhase = "accepted" + lifecycleRunning lifecyclePhase = "running" + lifecycleCompleted lifecyclePhase = "completed" + lifecycleFailed lifecyclePhase = "failed" + lifecycleRecovery lifecyclePhase = "recovery-required" + lifecycleIntent lifecyclePhase = "intent" + lifecyclePrepared lifecyclePhase = "prepared" + lifecycleApplied lifecyclePhase = "applied" +) + type Generation struct { Number uint64 Receipt packages.PackageReceipt @@ -39,5 +54,7 @@ type OperationStatus struct { OperationID, AddonID, Phase string Generation uint64 // RuntimeState is an observation, never a durable execution receipt. + Stage string `json:"stage,omitempty"` + FailureCode string `json:"failureCode,omitempty"` RuntimeState string `json:"runtimeState,omitempty"` } diff --git a/core/services/addons/upgrade_operation.go b/core/services/addons/upgrade_operation.go new file mode 100644 index 000000000..05cb5c099 --- /dev/null +++ b/core/services/addons/upgrade_operation.go @@ -0,0 +1,149 @@ +// SPDX-License-Identifier: MIT +package addons + +import ( + "context" + "errors" + "github.com/mudler/LocalAI/pkg/addons/contract" + "github.com/mudler/xlog" +) + +type upgradeExecutionKey struct{} + +type upgradeSnapshot struct { + Command LifecycleCommand + Actor contract.Actor + Status OperationStatus +} + +// SubmitUpgrade accepts one owner-managed upgrade. There is deliberately no +// queue: contention is a conflict, and replay only observes the original ID. +func (m *Manager) SubmitUpgrade(ctx context.Context, actor contract.Actor, cmd LifecycleCommand) (OperationStatus, error) { + if err := ctx.Err(); err != nil { + return OperationStatus{}, err + } + if m.admin == nil || m.admin(ctx, actor) != nil { + return OperationStatus{}, ErrForbidden + } + if cmd.Action != "upgrade" || !addonID.MatchString(cmd.AddonID) || cmd.OperationID == "" || len(cmd.OperationID) > 256 { + return OperationStatus{}, ErrConflict + } + m.upgradeMu.Lock() + defer m.upgradeMu.Unlock() + if m.upgradeClosing { + return OperationStatus{}, ErrNotOwner + } + if old, ok := m.upgradeViews[cmd.OperationID]; ok { + if old.Command != cmd || old.Actor != actor { + return OperationStatus{}, ErrConflict + } + return old.Status, nil + } + // Identify the same pinned transition from immutable projections even + // while execution owns mu; callers receive the same typed conflict then. + for _, old := range m.upgradeViews { + if cmd.ExpectedDigest != "" && old.Command.Action == "upgrade" && old.Command.AddonID == cmd.AddonID && old.Command.ExpectedDigest == cmd.ExpectedDigest && old.Command.CandidateDigest == cmd.CandidateDigest && old.Status.Phase != string(lifecycleFailed) { + return OperationStatus{}, ErrTransitionConflict + } + } + if m.upgradeActive || !m.mu.TryLock() { + return OperationStatus{}, ErrConflict + } + defer m.mu.Unlock() + if e := m.check(ctx); e != nil { + return OperationStatus{}, e + } + if m.uncertain || m.runtimeRecoveryRequired { + return OperationStatus{}, ErrRecoveryRequired + } + if old, ok := m.journal.Operations[cmd.OperationID]; ok { + if old.Command != cmd || old.Actor != actor { + return OperationStatus{}, ErrConflict + } + return old.Status, nil + } + if !m.journal.Addons[cmd.AddonID].Installed { + return OperationStatus{}, ErrConflict + } + if (cmd.ExpectedDigest != "" || cmd.CandidateDigest != "") && (cmd.ExpectedDigest == "" || cmd.CandidateDigest == "") { + return OperationStatus{}, ErrConflict + } + for _, op := range m.journal.Operations { + if cmd.ExpectedDigest != "" && op.Command.Action == "upgrade" && op.Command.AddonID == cmd.AddonID && op.Command.ExpectedDigest == cmd.ExpectedDigest && op.Command.CandidateDigest == cmd.CandidateDigest && op.Status.Phase != string(lifecycleFailed) { + return OperationStatus{}, ErrTransitionConflict + } + if op.Command.AddonID == cmd.AddonID && op.Status.Phase != string(lifecycleCompleted) && op.Status.Phase != string(lifecycleFailed) { + return OperationStatus{}, ErrConflict + } + } + // Preserve typed precondition refusals without accepting invalid work. The + // worker rechecks the pins before effects, since acceptance is not a promise + // that a mutable library candidate will stay unchanged. + if cmd.ExpectedDigest != "" { + if _, err := m.checkPins(ctx, cmd, m.journal.Addons[cmd.AddonID]); err != nil { + return OperationStatus{}, err + } + } + op := journalOperation{Command: cmd, Actor: actor, Async: true, Status: OperationStatus{OperationID: cmd.OperationID, AddonID: cmd.AddonID, Phase: string(lifecycleAccepted)}} + m.journal.Operations[cmd.OperationID] = op + // persist normally refreshes snapshots. We already hold the projection lock. + if e := m.persistJournal(string(lifecycleAccepted)); e != nil { + return OperationStatus{}, e + } + m.upgradeViews[cmd.OperationID] = upgradeSnapshot{cmd, actor, op.Status} + m.upgradeActive = true + m.upgradeWG.Add(1) + go func() { + defer m.upgradeWG.Done() + _, err := m.submit(context.WithValue(m.upgradeCtx, upgradeExecutionKey{}, m), actor, cmd, true) + 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.FailureCode = "upgrade_failed" + if errors.Is(err, context.Canceled) { + current.Status.FailureCode = "interrupted" + } + if m.dataBlocked(cmd.AddonID) { + current.Status.FailureCode = "recovery_required" + } + m.journal.Operations[cmd.OperationID] = current + _ = m.persist(string(lifecycleFailed)) + } + m.mu.Unlock() + if err != nil { + xlog.Warn("Owner upgrade ended", "operationId", cmd.OperationID, "error", err) + } + m.upgradeMu.Lock() + m.upgradeActive = false + m.upgradeMu.Unlock() + }() + return op.Status, nil +} + +// 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() { + 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} + } + } +} + +func (m *Manager) upgradeStage(id, stage string) error { + op := m.journal.Operations[id] + if !op.Async { + return nil + } + op.Status.Stage = stage + m.journal.Operations[id] = op + return m.persist(stage) +} diff --git a/core/services/addons/upgrade_operation_test.go b/core/services/addons/upgrade_operation_test.go new file mode 100644 index 000000000..74f564210 --- /dev/null +++ b/core/services/addons/upgrade_operation_test.go @@ -0,0 +1,290 @@ +// SPDX-License-Identifier: MIT +package addons + +import ( + "context" + "fmt" + "os" + "path/filepath" + "time" + + "github.com/mudler/LocalAI/core/services/addons/recovery" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// The policy export happens inside the real verified-backup path, before +// Installer.Commit. Cancelling here reproduces a caller going away while a +// backup completes, without relying on a large directory or a wall-clock race. +type upgradePolicy struct { + exporting func() + exportingContext func(context.Context) +} + +func (*upgradePolicy) Mode() string { return "named-user" } +func (*upgradePolicy) Hold(context.Context, string) (func(), error) { return func() {}, nil } +func (p *upgradePolicy) Export(ctx context.Context, _ string) (recovery.PolicyExport, error) { + if p.exportingContext != nil { + p.exportingContext(ctx) + } + if p.exporting != nil { + p.exporting() + } + return recovery.PolicyExport{}, nil +} +func (*upgradePolicy) Restore(context.Context, string, recovery.PolicyExport, recovery.RestoreOptions) (recovery.RestoreReport, error) { + return recovery.RestoreReport{}, nil +} +func (*upgradePolicy) LostWrites(context.Context, string, time.Time) (int, error) { return 0, nil } +func (*upgradePolicy) Forget(context.Context, string) (int, error) { return 0, nil } + +var _ = Describe("owner-managed upgrade operations", func() { + It("finishes a real backup and commit after the submitting request is cancelled", func() { + cfg, c, i := fixture() + m := open(cfg, c, i) + _, e := m.Submit(ctx, actor, command("initial", "install")) + Expect(e).NotTo(HaveOccurred()) + dir := filepath.Join(cfg.Root, "data", "sample") + Expect(os.MkdirAll(dir, 0700)).To(Succeed()) + Expect(os.WriteFile(filepath.Join(dir, "retained"), []byte("last good data"), 0600)).To(Succeed()) + request, cancel := context.WithCancel(ctx) + defer cancel() + exported := make(chan struct{}) + r, e := recovery.New(recovery.Config{Root: cfg.Root, Host: m.RecoveryHost(), Policy: &upgradePolicy{exporting: func() { close(exported); cancel() }}, Admin: admin}) + Expect(e).NotTo(HaveOccurred()) + m.SetRecovery(r) + result, e := m.SubmitUpgrade(request, actor, command("upgrade-disconnect", "upgrade")) + Expect(e).NotTo(HaveOccurred()) + Expect(result.Phase).To(Equal(string(lifecycleAccepted))) + Eventually(exported).Should(BeClosed()) + Eventually(func() string { + status, err := m.OperationStatus(ctx, actor, "upgrade-disconnect") + Expect(err).NotTo(HaveOccurred()) + result = status + return status.Phase + }).Should(Equal(string(lifecycleCompleted))) + Expect(result.Generation).To(Equal(uint64(2))) + }) + It("polls and retries during a blocked backup, rejects conflicting IDs, and retains the receipt on restart", func() { + cfg, c, i := fixture() + m := open(cfg, c, i) + _, e := m.Submit(ctx, actor, command("initial", "install")) + Expect(e).NotTo(HaveOccurred()) + dir := filepath.Join(cfg.Root, "data", "sample") + Expect(os.MkdirAll(dir, 0700)).To(Succeed()) + Expect(os.WriteFile(filepath.Join(dir, "retained"), []byte("data"), 0600)).To(Succeed()) + entered, release := make(chan struct{}), make(chan struct{}) + defer func() { + select { + case <-release: + default: + close(release) + } + }() + r, e := recovery.New(recovery.Config{Root: cfg.Root, Host: m.RecoveryHost(), Policy: &upgradePolicy{exporting: func() { close(entered); <-release }}, Admin: admin}) + Expect(e).NotTo(HaveOccurred()) + m.SetRecovery(r) + cmd := command("lost-reply", "upgrade") + _, e = m.SubmitUpgrade(ctx, actor, cmd) + Expect(e).NotTo(HaveOccurred()) + Eventually(entered).Should(BeClosed()) + read, cancel := context.WithTimeout(ctx, 100*time.Millisecond) + defer cancel() + status, e := m.OperationStatus(read, actor, cmd.OperationID) + Expect(e).NotTo(HaveOccurred()) + Expect(status.Stage).To(Equal("backing-up")) + status, e = m.SubmitUpgrade(read, actor, cmd) + Expect(e).NotTo(HaveOccurred()) + Expect(status.Phase).To(Equal(string(lifecycleRunning))) + changed := cmd + changed.Version = "9.0.0" + _, e = m.SubmitUpgrade(read, actor, changed) + Expect(e).To(MatchError(ErrConflict)) + changed = cmd + changed.OperationID = "another" + _, e = m.SubmitUpgrade(read, actor, changed) + Expect(e).To(MatchError(ErrConflict)) + other := actor + other.CredentialID = "expired" + _, e = m.OperationStatus(read, other, cmd.OperationID) + Expect(e).To(HaveOccurred()) + close(release) + Eventually(func() string { + st, e := m.OperationStatus(ctx, actor, cmd.OperationID) + Expect(e).NotTo(HaveOccurred()) + return st.Phase + }).Should(Equal(string(lifecycleCompleted))) + Expect(m.Close()).To(Succeed()) + second := open(cfg, c, i) + Expect(second.Reconcile(ctx)).To(Succeed()) + status, e = second.SubmitUpgrade(ctx, actor, cmd) + Expect(e).NotTo(HaveOccurred()) + Expect(status.Phase).To(Equal(string(lifecycleCompleted))) + Expect(status.Generation).To(Equal(uint64(2))) + }) + It("reconciles durable accepted work to interrupted without replay", func() { + cfg, c, i := fixture() + m := open(cfg, c, i) + _, e := m.Submit(ctx, actor, command("initial", "install")) + Expect(e).NotTo(HaveOccurred()) + cmd := command("interrupted", "upgrade") + m.mu.Lock() + m.journal.Operations[cmd.OperationID] = journalOperation{Command: cmd, Actor: actor, Async: true, Status: OperationStatus{OperationID: cmd.OperationID, AddonID: cmd.AddonID, Phase: string(lifecycleAccepted)}} + Expect(m.persist(string(lifecycleAccepted))).To(Succeed()) + m.mu.Unlock() + Expect(m.Close()).To(Succeed()) + second := open(cfg, c, i) + Expect(second.Reconcile(ctx)).To(Succeed()) + status, e := second.OperationStatus(ctx, actor, cmd.OperationID) + Expect(e).NotTo(HaveOccurred()) + Expect(status.Phase).To(Equal(string(lifecycleFailed))) + Expect(status.FailureCode).To(Equal("interrupted")) + st, e := second.Status(ctx, "sample") + Expect(e).NotTo(HaveOccurred()) + Expect(st.Generation).To(Equal(uint64(1))) + }) + It("Close cancels and joins an accepted owner worker before releasing storage", func() { + cfg, c, i := fixture() + m := open(cfg, c, i) + _, e := m.Submit(ctx, actor, command("initial", "install")) + Expect(e).NotTo(HaveOccurred()) + // Hold the execution gate to ensure Close must join the accepted worker. + m.opMu.Lock() + _, e = m.SubmitUpgrade(ctx, actor, command("shutdown", "upgrade")) + Expect(e).NotTo(HaveOccurred()) + done := make(chan error, 1) + go func() { done <- m.Close() }() + Consistently(done, 30*time.Millisecond).ShouldNot(Receive()) + m.opMu.Unlock() + Eventually(done).Should(Receive(BeNil())) + second := open(cfg, c, i) + Expect(second.Reconcile(ctx)).To(Succeed()) + status, e := second.OperationStatus(ctx, actor, "shutdown") + Expect(e).NotTo(HaveOccurred()) + Expect(status.Phase).To(Equal(string(lifecycleFailed))) + Expect(status.FailureCode).To(Equal("interrupted")) + }) + + It("Close cancels a backup in flight and joins its cleanup before releasing ownership", func() { + cfg, c, i := fixture() + m := open(cfg, c, i) + _, e := m.Submit(ctx, actor, command("initial", "install")) + Expect(e).NotTo(HaveOccurred()) + dir := filepath.Join(cfg.Root, "data", "sample") + Expect(os.MkdirAll(dir, 0700)).To(Succeed()) + Expect(os.WriteFile(filepath.Join(dir, "retained"), []byte("last good data"), 0600)).To(Succeed()) + entered, cancelled, release := make(chan struct{}), make(chan struct{}), make(chan struct{}) + defer func() { + select { + case <-release: + default: + close(release) + } + }() + r, e := recovery.New(recovery.Config{Root: cfg.Root, Host: m.RecoveryHost(), Policy: &upgradePolicy{exportingContext: func(work context.Context) { close(entered); <-work.Done(); close(cancelled); <-release }}, Admin: admin}) + Expect(e).NotTo(HaveOccurred()) + m.SetRecovery(r) + _, e = m.SubmitUpgrade(ctx, actor, command("shutdown-backup", "upgrade")) + Expect(e).NotTo(HaveOccurred()) + Eventually(entered).Should(BeClosed()) + done := make(chan error, 1) + go func() { done <- m.Close() }() + Eventually(cancelled).Should(BeClosed()) + // Cancellation reached real backup I/O, but its callback has not returned. + select { + case e := <-done: + Fail("Close returned before backup cleanup: " + fmt.Sprint(e)) + default: + } + _, e = NewManager(ctx, cfg, c, i, admin) + Expect(e).To(MatchError(ErrOwnerBusy)) + close(release) + Eventually(done).Should(Receive(BeNil())) + second := open(cfg, c, i) + Expect(second.Reconcile(ctx)).To(Succeed()) + status, e := second.OperationStatus(ctx, actor, "shutdown-backup") + Expect(e).NotTo(HaveOccurred()) + Expect(status.Phase).To(Equal(string(lifecycleFailed))) + st, e := second.Status(ctx, "sample") + Expect(e).NotTo(HaveOccurred()) + Expect(st.Generation).To(Equal(uint64(1))) + data, e := os.ReadFile(filepath.Join(dir, "retained")) + Expect(e).NotTo(HaveOccurred()) + Expect(string(data)).To(Equal("last good data")) + }) + It("serializes synchronous disable against an accepted upgrade without overwriting its journal", func() { + // An accepted journal loaded before restart reconciliation has no worker. + // This pins the guard itself independently of opMu waiter ordering below. + cfgPending, cPending, iPending := fixture() + pending := open(cfgPending, cPending, iPending) + _, err := pending.Submit(ctx, actor, command("initial", "install")) + Expect(err).NotTo(HaveOccurred()) + acceptedCmd := command("pending", "upgrade") + pending.mu.Lock() + pending.journal.Operations[acceptedCmd.OperationID] = journalOperation{Command: acceptedCmd, Actor: actor, Async: true, Status: OperationStatus{OperationID: acceptedCmd.OperationID, AddonID: acceptedCmd.AddonID, Phase: string(lifecycleAccepted)}} + Expect(pending.persist(string(lifecycleAccepted))).To(Succeed()) + pending.mu.Unlock() + Expect(pending.Close()).To(Succeed()) + loaded := open(cfgPending, cPending, iPending) + _, err = loaded.Submit(ctx, actor, command("must-not-overtake", "disable")) + Expect(err).To(MatchError(ErrRecoveryRequired)) + loaded.mu.Lock() + Expect(loaded.journal.Operations[acceptedCmd.OperationID].Status.Phase).To(Equal(string(lifecycleAccepted))) + _, exists := loaded.journal.Operations["must-not-overtake"] + Expect(exists).To(BeFalse()) + Expect(loaded.journal.Addons["sample"].Generation).To(Equal(uint64(1))) + loaded.mu.Unlock() + + cfg, c, i := fixture() + m := open(cfg, c, i) + _, e := m.Submit(ctx, actor, command("initial", "install")) + Expect(e).NotTo(HaveOccurred()) + m.opMu.Lock() + locked := true + defer func() { + if locked { + m.opMu.Unlock() + } + }() + cmd := command("accepted-disable", "upgrade") + _, e = m.SubmitUpgrade(ctx, actor, cmd) + Expect(e).NotTo(HaveOccurred()) + status, e := m.OperationStatus(ctx, actor, cmd.OperationID) + Expect(e).NotTo(HaveOccurred()) + Expect(status.Phase).To(Equal(string(lifecycleAccepted))) + done := make(chan error, 1) + started := make(chan struct{}) + go func() { + close(started) + _, err := m.Submit(ctx, actor, command("concurrent-disable", "disable")) + done <- err + }() + <-started + m.opMu.Unlock() + locked = false + var submitErr error + Eventually(done).Should(Receive(&submitErr)) + // Either the journal guard rejects the unresolved acceptance, or opMu runs + // disable after the upgrade completes. Neither ordering may overtake it. + if submitErr != nil { + Expect(submitErr).To(MatchError(ErrRecoveryRequired)) + } + Eventually(func() string { + st, e := m.OperationStatus(ctx, actor, cmd.OperationID) + Expect(e).NotTo(HaveOccurred()) + return st.Phase + }).Should(Equal(string(lifecycleCompleted))) + m.mu.Lock() + defer m.mu.Unlock() + Expect(m.journal.Operations[cmd.OperationID].Command).To(Equal(cmd)) + Expect(m.journal.Addons["sample"].Generation).To(Equal(uint64(2))) + if submitErr != nil { + _, exists := m.journal.Operations["concurrent-disable"] + Expect(exists).To(BeFalse()) + } else { + Expect(m.journal.Operations["concurrent-disable"].Status.Phase).To(Equal(string(lifecycleCompleted))) + Expect(m.journal.Addons["sample"].DesiredEnabled).To(BeFalse()) + } + }) + +}) diff --git a/core/services/addons/upgrade_state.go b/core/services/addons/upgrade_state.go new file mode 100644 index 000000000..bb7bff763 --- /dev/null +++ b/core/services/addons/upgrade_state.go @@ -0,0 +1,40 @@ +// SPDX-License-Identifier: MIT +package addons + +// 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. +type upgradeEvent string + +const ( + upgradeStart upgradeEvent = "start" + upgradeComplete upgradeEvent = "complete" + upgradeFail upgradeEvent = "fail" + upgradeInterrupt upgradeEvent = "interrupt" + upgradeRecover upgradeEvent = "recover" + upgradeDetach upgradeEvent = "detach" + upgradeRetry upgradeEvent = "retry" +) + +func upgradeTransition(phase lifecyclePhase, event upgradeEvent) (next lifecyclePhase, rule string) { + switch phase { + case lifecycleCompleted, lifecycleFailed, lifecycleRecovery: + return phase, "U0-terminal" + case lifecycleAccepted, lifecycleRunning: + switch event { + case upgradeDetach, upgradeRetry: + return phase, "U1-observe" + case upgradeStart: + return lifecycleRunning, "U2-start" + case upgradeComplete: + return lifecycleCompleted, "U3-complete" + case upgradeFail: + return lifecycleFailed, "U4-fail" + case upgradeInterrupt: + return lifecycleFailed, "U5-interrupted" + case upgradeRecover: + return lifecycleRecovery, "U6-recovery" + } + } + panic("invalid upgrade state or event") +} diff --git a/core/services/addons/upgrade_state_test.go b/core/services/addons/upgrade_state_test.go new file mode 100644 index 000000000..f649368eb --- /dev/null +++ b/core/services/addons/upgrade_state_test.go @@ -0,0 +1,52 @@ +// SPDX-License-Identifier: MIT +package addons + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("upgrade observation reducer", func() { + It("traces completion, retries, detach, interruption and late delivery without replay", func() { + for _, trace := range []struct { + events []upgradeEvent + rules []string + end lifecyclePhase + }{ + {[]upgradeEvent{upgradeStart, upgradeComplete}, []string{"U2-start", "U3-complete"}, lifecycleCompleted}, + {[]upgradeEvent{upgradeRetry, upgradeStart, upgradeRetry, upgradeComplete}, []string{"U1-observe", "U2-start", "U1-observe", "U3-complete"}, lifecycleCompleted}, + {[]upgradeEvent{upgradeStart, upgradeDetach, upgradeComplete}, []string{"U2-start", "U1-observe", "U3-complete"}, lifecycleCompleted}, + {[]upgradeEvent{upgradeInterrupt, upgradeStart}, []string{"U5-interrupted", "U0-terminal"}, lifecycleFailed}, + {[]upgradeEvent{upgradeRecover, upgradeComplete}, []string{"U6-recovery", "U0-terminal"}, lifecycleRecovery}, + } { + phase := lifecycleAccepted + for n, event := range trace.events { + var rule string + phase, rule = upgradeTransition(phase, event) + Expect(rule).To(Equal(trace.rules[n])) + } + Expect(phase).To(Equal(trace.end)) + } + }) + It("keeps terminals immutable for every event and gives each active state a terminal exit", 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) + Expect(next).NotTo(BeEmpty()) + Expect(rule).NotTo(BeEmpty()) + } + } + for _, phase := range []lifecyclePhase{lifecycleCompleted, lifecycleFailed, lifecycleRecovery} { + for _, event := range events { + next, _ := upgradeTransition(phase, event) + Expect(next).To(Equal(phase)) + } + } + for _, phase := range []lifecyclePhase{lifecycleAccepted, lifecycleRunning} { + next, rule := upgradeTransition(phase, upgradeInterrupt) + Expect(next).To(Equal(lifecycleFailed)) + Expect(rule).To(Equal("U5-interrupted")) + } + }) +}) diff --git a/docs/content/development/addon-testing.md b/docs/content/development/addon-testing.md index 7032268df..9533977d8 100644 --- a/docs/content/development/addon-testing.md +++ b/docs/content/development/addon-testing.md @@ -461,3 +461,22 @@ The picker specs exercise rendering and switches; they do not alone establish re Also test actual provider resolution, grants, pins, OFF persistence, and index/load agreement against the application. Require native linux/amd64 and linux/arm64 execution evidence before claiming release verification. Cross-compilation or a configured CI matrix alone does not establish native execution. + +## Upgrade submission and observation regressions + +- [ ] `go test -race ./core/services/addons ./core/services/addons/gateway ./core/services/addons/recovery -count=1` +- [ ] `cd core/http/react-ui && node --test tests/unit/updatereview.test.mjs` +- [ ] `cd core/http/react-ui && npm run build` +- [ ] Verify cancellation during the real verified-backup path does not cancel + an accepted upgrade or prevent its package commit. +- [ ] Verify status reads and same-ID retries through two real mTLS frontends + remain available while backup exceeds an injected ordinary client deadline. +- [ ] Verify payload conflicts, actor isolation, terminal receipt replay after + restart, interruption without replay, and Close cancellation/join behavior. +- [ ] Verify browser reload observes the stored ID, account switching isolates + records, logout clears records, and progress does not invent percentages. + +The recovery suite above remains required for crash boundaries during backup, +staging, migration and publication, including format-changing recovery-required +outcomes. The focused submission tests do not replace the real executable, +real-auth HTTP or rendered-browser acceptance gates. diff --git a/docs/content/features/addons.md b/docs/content/features/addons.md index 705ed8865..5f738fa69 100644 --- a/docs/content/features/addons.md +++ b/docs/content/features/addons.md @@ -4446,3 +4446,41 @@ What the source tree does not prove: the images and the `runsc` download are not Add-on skill switches and tool-provider choices read the owner policy and update only the changed provider keys. Clearing a choice removes that pin without changing other add-ons. Revision conflicts retain edits for an explicit reload and rebase. Uncertain writes retry the same operation; an owner outage disables writes until reconciliation. Native add-on configuration remains separate. With add-ons enabled, Settings → Backends uses the owner policy for automatic upgrades rather than the frontend runtime file. Combined mode uses the embedded owner through the same API. + +### Upgrade execution and polling + +Browser upgrades are accepted durably by the add-on owner before the submission +returns. The response identifies the same operation used by subsequent status +reads. Closing a page, refreshing, or losing the acceptance reply does not cancel +accepted work. Accepted upgrades cannot be cancelled from the UI. +The Activity entry has no Cancel control after submission. A retry with the same operation ID, actor and payload observes that +operation; a changed payload conflicts. Only one asynchronous upgrade is admitted +at a time; competing submissions receive a conflict instead of entering a queue. +Other lifecycle service callers retain their synchronous contract. + +Execution is single-owner, using the existing storage ownership fence and local +journal, not frontend memory or a message-bus dependency. Standalone uses the same +path. Any frontend can forward polling to the owner. Each read revalidates current +administrator authorization and the recorded actor, including credential identity. +Expired or revoked credentials cannot read the operation, but do not revoke work +already accepted. A different credential does not inherit the original receipt. + +Activity shows real stages (package preparation, verified backup, candidate +migration/health, and publication), not estimated byte percentages. Failed +operations retain a safe failure code; detailed underlying causes go to owner +logs. A migration failure can require explicit data recovery; keeping the old +package does not mean the data is safe to run. + +Shutdown cancels and waits for the owner worker before releasing storage. On +restart, existing publication and data-recovery journals reconcile before runtime +startup: prepared publication may finish; interrupted work otherwise becomes a +terminal interrupted failure, with recovery-required reported when data is blocked. +Destructive migration is never automatically replayed. An uncertain publication +remains fenced for reconciliation, not presented as a definite rollback. + +The browser remembers pending operation IDs in session storage, scoped to the +signed-in user, and resumes observation after a reload without resubmitting. +Logout clears the current user's observer records. Storage restrictions may prevent +observer recovery, but never cancel the owner operation. An unreadable owner or +exhausted polling window is an unknown outcome: keep the ID and read it again, +rather than starting a replacement operation. diff --git a/scripts/addons/suites.json b/scripts/addons/suites.json index da6028acc..604381df4 100644 --- a/scripts/addons/suites.json +++ b/scripts/addons/suites.json @@ -490,7 +490,15 @@ "needs a state root for the owner only", "starts the enabled add-ons side by side, so the start takes the slowest ones and not the sum", "distinguishes healthy empty connector discovery from loss of runtime ownership", - "installs upgrades and uninstalls the actual Sandbox document skills through verified receipts" + "installs upgrades and uninstalls the actual Sandbox document skills through verified receipts", + "finishes a real backup and commit after the submitting request is cancelled", + "polls and retries during a blocked backup, rejects conflicting IDs, and retains the receipt on restart", + "reconciles durable accepted work to interrupted without replay", + "Close cancels and joins an accepted owner worker before releasing storage", + "traces completion, retries, detach, interruption and late delivery without replay", + "keeps terminals immutable for every event and gives each active state a terminal exit", + "Close cancels a backup in flight and joins its cleanup before releasing ownership", + "serializes synchronous disable against an accepted upgrade without overwriting its journal" ], "report": ".cache/addons/logs/lifecycle.json" }, @@ -803,7 +811,8 @@ "projects private policy safely", "forwards authenticated owner decisions", "validates strict request payloads", - "checks CAS and administrator authority" + "checks CAS and administrator authority", + "accepts before a slow backup, detaches the request and polls through either frontend" ], "report": ".cache/addons/logs/gateway.json", "reports": [ @@ -911,7 +920,8 @@ "projects private policy safely", "forwards authenticated owner decisions", "validates strict request payloads", - "checks CAS and administrator authority" + "checks CAS and administrator authority", + "accepts before a slow backup, detaches the request and polls through either frontend" ] }, { @@ -2027,7 +2037,8 @@ "projects private policy safely", "forwards authenticated owner decisions", "validates strict request payloads", - "checks CAS and administrator authority" + "checks CAS and administrator authority", + "accepts before a slow backup, detaches the request and polls through either frontend" ], "report": ".cache/addons/logs/artifacts.json", "reports": [ @@ -2177,7 +2188,8 @@ "projects private policy safely", "forwards authenticated owner decisions", "validates strict request payloads", - "checks CAS and administrator authority" + "checks CAS and administrator authority", + "accepts before a slow backup, detaches the request and polls through either frontend" ] }, { @@ -2590,7 +2602,15 @@ "records the chat with the envelope of a read and of an action while it runs", "sends the add-on an envelope without a new field", "distinguishes healthy empty connector discovery from loss of runtime ownership", - "installs upgrades and uninstalls the actual Sandbox document skills through verified receipts" + "installs upgrades and uninstalls the actual Sandbox document skills through verified receipts", + "finishes a real backup and commit after the submitting request is cancelled", + "polls and retries during a blocked backup, rejects conflicting IDs, and retains the receipt on restart", + "reconciles durable accepted work to interrupted without replay", + "Close cancels and joins an accepted owner worker before releasing storage", + "traces completion, retries, detach, interruption and late delivery without replay", + "keeps terminals immutable for every event and gives each active state a terminal exit", + "Close cancels a backup in flight and joins its cleanup before releasing ownership", + "serializes synchronous disable against an accepted upgrade without overwriting its journal" ], "report": ".cache/addons/logs/apps.json", "reports": [ @@ -2681,7 +2701,15 @@ "keeps the manual upgrade closed to a caller who is not an administrator", "starts the enabled add-ons side by side, so the start takes the slowest ones and not the sum", "distinguishes healthy empty connector discovery from loss of runtime ownership", - "installs upgrades and uninstalls the actual Sandbox document skills through verified receipts" + "installs upgrades and uninstalls the actual Sandbox document skills through verified receipts", + "finishes a real backup and commit after the submitting request is cancelled", + "polls and retries during a blocked backup, rejects conflicting IDs, and retains the receipt on restart", + "reconciles durable accepted work to interrupted without replay", + "Close cancels and joins an accepted owner worker before releasing storage", + "traces completion, retries, detach, interruption and late delivery without replay", + "keeps terminals immutable for every event and gives each active state a terminal exit", + "Close cancels a backup in flight and joins its cleanup before releasing ownership", + "serializes synchronous disable against an accepted upgrade without overwriting its journal" ] }, { diff --git a/tests/addons/foundation/recovery_failure_test.go b/tests/addons/foundation/recovery_failure_test.go index 7f781c900..df0af24c2 100644 --- a/tests/addons/foundation/recovery_failure_test.go +++ b/tests/addons/foundation/recovery_failure_test.go @@ -23,7 +23,28 @@ var _ = Describe("failed starts and interrupted lifecycle", func() { admin := c.Token(cluster.FrontB, "admin") marker := retainData(c, "found-tool") r := lifecycle(admin, "found-tool", "upgrade", 0, "2.0.0") - Expect(r.Status).NotTo(Equal(200), "an upgrade whose process cannot start must not report success: "+r.String()) + Expect(r.Status).To(Equal(200), r.String()) + var accepted struct { + Payload struct{ OperationID, Phase string } + } + Expect(r.JSON(&accepted)).To(Succeed()) + Expect(accepted.Payload.Phase).To(Equal("accepted")) + settle("the accepted upgrade reporting terminal failure", 30*time.Second, func() bool { + status := dispatch(admin, map[string]any{"operationId": accepted.Payload.OperationID, "addonId": "found-tool", "action": "operation-status", "arguments": map[string]any{}}) + Expect(status.Status).To(Equal(200), status.String()) + var out struct { + Payload struct { + Phase string + FailureCode string `json:"failureCode"` + } + } + Expect(status.JSON(&out)).To(Succeed()) + if out.Payload.Phase != "failed" { + return false + } + Expect(out.Payload.FailureCode).To(Equal("upgrade_failed")) + return true + }) a := appOf(admin, "found-tool") Expect(a).NotTo(BeNil()) Expect(a.State).To(Equal("running")) diff --git a/tests/addons/tablesacceptance/replicas_test.go b/tests/addons/tablesacceptance/replicas_test.go index 8f4b3feb6..67139cabc 100644 --- a/tests/addons/tablesacceptance/replicas_test.go +++ b/tests/addons/tablesacceptance/replicas_test.go @@ -434,6 +434,23 @@ var _ = Describe("Tables upgrade with an explicit check between two packaged ver op := opID("up") done := upgrade(cluster.FrontB, op, installed, candidate) Expect(done.Status).To(Equal(200), done.String()) + Expect(payload(done)).To(HaveKeyWithValue("Phase", "accepted")) + Expect(payload(done)).To(HaveKeyWithValue("OperationID", op)) + // A running old generation is not evidence that accepted work finished. + // Observe the same actor-bound operation through the other frontend. + var terminal map[string]any + Eventually(func() string { + status := d.tab(cluster.FrontA, "admin", "POST", "/api/addons/dispatch", map[string]any{ + "addonId": "tables", "operationId": op, "action": "operation-status", "arguments": map[string]any{}, + }, "") + Expect(status.Status).To(Equal(200), status.String()) + terminal = payload(status) + Expect(terminal).To(HaveKeyWithValue("OperationID", op)) + phase, _ := terminal["Phase"].(string) + return phase + }, 30*time.Second, 100*time.Millisecond).Should(Or(Equal("completed"), Equal("failed"))) + Expect(terminal).To(HaveKeyWithValue("Phase", "completed")) + Expect(terminal).To(HaveKeyWithValue("Generation", float64(genBefore+1))) d.waitState("running") for _, n := range nodes { a := d.app(d.c.Token(n, "admin")) @@ -442,6 +459,7 @@ var _ = Describe("Tables upgrade with an explicit check between two packaged ver // The same operation from another node is the recorded result. again := upgrade(cluster.FrontA, op, installed, candidate) Expect(again.Status).To(Equal(200), again.String()) + Expect(payload(again)).To(Equal(terminal), "the same ID replays the terminal receipt without another upgrade") // Data, grants and the receipts survived, and the new generation works. for _, n := range nodes { Expect(d.rowCount(n, "member", res)).To(Equal(before), n)