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)