diff --git a/backend/drizzle/0054_mcp_geoip_retirement.sql b/backend/drizzle/0054_mcp_geoip_retirement.sql new file mode 100644 index 00000000..6b59b470 --- /dev/null +++ b/backend/drizzle/0054_mcp_geoip_retirement.sql @@ -0,0 +1,18 @@ +BEGIN; +CREATE OR REPLACE FUNCTION public.mcp_ordinary_retirement_claim_immutable() RETURNS trigger +LANGUAGE plpgsql SET TimeZone='UTC' SET search_path=pg_catalog,public AS $$ +BEGIN + IF TG_OP='INSERT' AND NEW.claim_xid=pg_current_xact_id() AND EXISTS( + SELECT 1 FROM public.mcp_operations o JOIN public.mcp_plans p ON p.id=o.plan_id AND p.connection_id=o.connection_id + WHERE o.id=NEW.operation_id AND o.plan_id=NEW.plan_id AND o.connection_id=NEW.connection_id + AND o.state='applied' AND o.verification='passed' AND o.error_code IS NULL + AND isfinite(o.finished_at) AND o.finished_at createHash('sha256').update(buffer).digest('hex') async function compile(sourceDirectory) { for (const entry of await readdir(sourceDirectory, { withFileTypes: true })) { @@ -59,7 +60,7 @@ try { // embedded-postgres waits for a new exit event; avoid waiting again if its // owned child has already exited. This never targets another database. if (postgres.process?.exitCode === null) await postgres.stop() - if (dirname(resolve(directory)) !== resolve(root, 'tmp') || !directory.includes('routing-controller-baseline-owned-')) throw new Error('Owned routing cleanup boundary changed') + if (dirname(resolve(directory)) !== resolve(root, 'tmp') || !basename(directory).startsWith(ownedPrefix)) throw new Error('Owned routing cleanup boundary changed') await rm(join(directory, 'pg'), { recursive: true, force: true, maxRetries: 20, retryDelay: 100 }) console.log('Owned routing PostgreSQL stopped and its data removed; snapshot/log evidence retained') } diff --git a/backend/src/db/migration-manifest.test.ts b/backend/src/db/migration-manifest.test.ts index e5757324..89a96a0b 100644 --- a/backend/src/db/migration-manifest.test.ts +++ b/backend/src/db/migration-manifest.test.ts @@ -12,10 +12,10 @@ const reviewed = (source: string, extra: Partial = {}) => { const bytes = Buffer.from(source), file = "9000_fixture.sql" return () => analyzeMigration(file, bytes, { file, sha256: createHash("sha256").update(bytes).digest("hex"), transaction: "owner-transaction", selfMarker: false, opaqueBodies: 0, bodyReview: [], effects: ["Explicit inert fixture review"], ...extra }) } -await test("all 53 exact file pins and every top-level schema operation are inventoried", () => { - assert.equal(inventory.length, 53); assert.equal(MIGRATION_MANIFEST.length, 53) +await test("all 55 exact file pins and every top-level schema operation are inventoried", () => { + assert.equal(inventory.length, 55); assert.equal(MIGRATION_MANIFEST.length, 55) assert.equal(inventory.filter(file => file.transaction === "owner-transaction").length, 17) - assert.equal(inventory.filter(file => file.transaction === "self-managed-single-transaction").length, 36) + assert.equal(inventory.filter(file => file.transaction === "self-managed-single-transaction").length, 38) assert(inventory.every(file => file.operations.length === file.statementCount && file.effects.length > 0)) let tables = new Set() for (const file of inventory) { @@ -32,8 +32,8 @@ await test("DO BEGIN and stored function BEGIN are never transaction boundaries" assert.equal(endpoint.topLevelTransactionControls.length, 2); assert.equal(endpoint.opaqueBodies.length, 2) assert.deepEqual(endpoint.opaqueBodies.map(body => body.kind), ["stored-function-body", "executed-do-body"]) }) -await test("0035 through 0053 have exact atomic file-owned terminal markers", () => { - assert.deepEqual(inventory.filter(file => file.selfMarker).map(file => file.id), ["0035_server_probe_journal", "0036_diagnostic_cancellations", "0037_server_probe_storage_v2", "0038_server_poll_journal_and_writer_guards", "0039_server_poll_storage", "0040_mcp_routing_reservations", "0041_server_poll_terminal_archives", "0042_server_poll_settings_epoch", "0043_mcp_routing_forward_charges", "0044_mcp_routing_manual_inverse_charges", "0045_mcp_operation_replay_tombstones", "0046_server_poll_operation_retirement", "0047_server_poll_retirement_claims", "0048_mcp_ordinary_retirement_claims", "0049_mcp_ordinary_telemetry_retirement", "0050_server_probe_retired_sources", "0051_server_probe_retirement_claims", "0052_server_probe_retired_gc", "0053_server_probe_terminal_proofs"]) +await test("0035 through 0055 have exact atomic file-owned terminal markers", () => { + assert.deepEqual(inventory.filter(file => file.selfMarker).map(file => file.id), ["0035_server_probe_journal", "0036_diagnostic_cancellations", "0037_server_probe_storage_v2", "0038_server_poll_journal_and_writer_guards", "0039_server_poll_storage", "0040_mcp_routing_reservations", "0041_server_poll_terminal_archives", "0042_server_poll_settings_epoch", "0043_mcp_routing_forward_charges", "0044_mcp_routing_manual_inverse_charges", "0045_mcp_operation_replay_tombstones", "0046_server_poll_operation_retirement", "0047_server_poll_retirement_claims", "0048_mcp_ordinary_retirement_claims", "0049_mcp_ordinary_telemetry_retirement", "0050_server_probe_retired_sources", "0051_server_probe_retirement_claims", "0052_server_probe_retired_gc", "0053_server_probe_terminal_proofs", "0054_mcp_geoip_retirement", "0055_mcp_backup_schedule_retirement"]) for (const file of inventory.filter(file => file.selfMarker)) { assert.equal(file.selfMarker!.atomicWithAllFileEffects, true); assert.equal(file.selfMarker!.statement, file.statementCount - 2); assert.equal(file.currentOuterMarkerGap, false) } assert.equal(inventory.filter(file => file.currentOuterMarkerGap).length, 34) }) diff --git a/backend/src/db/migration-manifest.ts b/backend/src/db/migration-manifest.ts index e9ccc324..9d4e5d32 100644 --- a/backend/src/db/migration-manifest.ts +++ b/backend/src/db/migration-manifest.ts @@ -828,6 +828,32 @@ const classifiedMigrationPins: readonly MigrationReview[] = [ "effects": [ "Creates empty bounded private terminal proof storage. Existing payload retention, scheduler, and archive GC remain unchanged." ] + }, + { + "file": "0054_mcp_geoip_retirement.sql", + "sha256": "680390f98076e3047dbb5a34eab417b253853703932d16c17e4fa3a11060accc", + "transaction": "self-managed-single-transaction", + "selfMarker": true, + "opaqueBodies": 1, + "bodyReview": [ + "Adds only exact settled GeoIP settings version 1 to the existing immutable ordinary retirement claim guard." + ], + "effects": [ + "Replaces the ordinary claim guard function without changing settings or retiring existing rows during migration." + ] + }, + { + "file": "0055_mcp_backup_schedule_retirement.sql", + "sha256": "07e63c1245ea547176d613933a81ad354e734fa64218dadced67ee8f965c01b6", + "transaction": "self-managed-single-transaction", + "selfMarker": true, + "opaqueBodies": 1, + "bodyReview": [ + "Adds only exact settled backup schedule settings version 1 to the existing immutable ordinary retirement claim guard." + ], + "effects": [ + "Replaces the ordinary claim guard function without changing backup settings, metadata or entries during migration." + ] } ] diff --git a/backend/src/modules/mcp/ordinary-operation-retirement.owned.test.ts b/backend/src/modules/mcp/ordinary-operation-retirement.owned.test.ts index c295ef6d..87cce853 100644 --- a/backend/src/modules/mcp/ordinary-operation-retirement.owned.test.ts +++ b/backend/src/modules/mcp/ordinary-operation-retirement.owned.test.ts @@ -16,6 +16,8 @@ const {tickMcpWorkers}=await import("./workers.js") const {pauseSchedulerJobsAndWait,resumeSchedulerJobs}=await import("../../services/scheduler-running.js") const {Client}=await import("pg") const {McpError}=await import("./security.js") +const {touchGeoipRunMeta}=await import("../../services/geoip-settings.js") +const {touchBackupScheduleRunMeta}=await import("../../services/backup-service.js") const errorIs=(code:string,status:number)=>(error:unknown)=>error instanceof McpError&&error.code===code&&error.statusCode===status try { @@ -25,19 +27,20 @@ try { const serverId=Number((await pool.query("INSERT INTO servers(name,host,enabled) VALUES('owned logs retention','127.0.0.1',false) RETURNING id")).rows[0].id) async function access(){ const actor={id:randomUUID(),email:"owned-logs-retirement@example.test",name:"Owned settings retention",apps:["mm"], - permissions:["mm:settings:admin","mm:mcp:approve","mm:traffic:write","mm:uptime:write","mm:servers:write","mm:network:write"], + permissions:["mm:settings:admin","mm:mcp:approve","mm:traffic:write","mm:uptime:write","mm:servers:write","mm:network:write","mm:backups:write"], issuedAt:Date.now()-1000,expiresAt:Date.now()+300000} const issued=await createMcpConnection(mcpConnectionCreateSchema.parse({name:actor.name,mode:"approval", permissions:actor.permissions,expiresAt:new Date(actor.expiresAt).toISOString(), scope:{serverIds:[serverId],operationIds:["logs.settings.update","traffic.settings.update","uptime.settings.update", - "servers.api-ping.settings.update","internet-path.settings.update","operations.read","changes.apply"],allowGlobal:true}, - limits:{callsPerMinute:100,mutationsPerHour:10}}),actor) + "servers.api-ping.settings.update","internet-path.settings.update","geoip.settings.update","backups.schedule.update","operations.read","changes.apply"],allowGlobal:true}, + limits:{callsPerMinute:100,mutationsPerHour:20}}),actor) return {actor,principal:await authenticateMcpToken(issued.token)} } const {actor,principal}=await access(),foreign=(await access()).principal async function operation(retentionDays:number,ageMs:number,tool="mm_logs_settings_update_plan", settings:Record={retentionDays}){ const plan=await invokeMcpTool(principal,tool,{settings}) as {planId:string;state:string} + if(tool==="mm_geoip_settings_update_plan"||tool==="mm_backup_schedule_update_plan")assert.equal(plan.state,"awaiting-approval") if(plan.state==="awaiting-approval")await approvePlan(plan.planId,actor,true) const key=randomUUID(),applied=await applyPlan(principal,plan.planId,key) let terminal @@ -141,8 +144,83 @@ try { await assert.rejects(applyPlan(principal,record.planId,record.key),errorIs("OPERATION_HISTORY_EXPIRED",410)) await assert.rejects(invokeMcpTool(foreign,"mm_operation_get",{operationId:record.id}),errorIs("NOT_FOUND",404)) } + // Keep the owned fixture offline even if a product scheduler refresh occurs. + await pool.query("UPDATE geoip_settings SET enabled=false WHERE id=1") + await touchGeoipRunMeta({lastCheckAt:new Date().toISOString(),lastError:"owned GeoIP metadata",etags:{"owned-test":"unchanged"}}) + const geoip=await operation(0,age+60000,"mm_geoip_settings_update_plan",{enabled:false,updateIntervalSec:86400}) + const geoipState=async()=>(await pool.query("SELECT to_jsonb(g) AS row FROM geoip_settings g WHERE id=1")).rows[0].row + const geoipBefore=await geoipState() + assert.equal(geoipBefore.enabled,false);assert.equal(geoipBefore.update_interval_sec,86400) + assert.equal(geoipBefore.last_error,"owned GeoIP metadata");assert.deepEqual(geoipBefore.etags_json,{"owned-test":"unchanged"}) + const {rows:[geoipStep]}=await pool.query("SELECT result FROM mcp_operation_steps WHERE operation_id=$1",[geoip.id]) + const assertGeoipUnclaimed=async()=>{ + assert.equal((await pool.query("SELECT 1 FROM mcp_ordinary_retirement_claims WHERE operation_id=$1",[geoip.id])).rows.length,0) + assert.equal((await pool.query("SELECT 1 FROM mcp_operations WHERE id=$1",[geoip.id])).rows.length,1) + assert.deepEqual(await geoipState(),geoipBefore) + } + await pool.query("UPDATE mcp_operation_steps SET result=jsonb_set(result,'{settings,updateIntervalSec}','172800') WHERE operation_id=$1",[geoip.id]) + await tickMcpWorkers();await tickMcpWorkers();await assertGeoipUnclaimed() + await pool.query("UPDATE mcp_operation_steps SET result=$2 WHERE operation_id=$1",[geoip.id,geoipStep.result]) + const geoipHold=randomUUID() + await pool.query("INSERT INTO mcp_drift(id,operation_id,expected,observed) VALUES($1,$2,'{}','{}')",[geoipHold,geoip.id]) + await tickMcpWorkers();await assertGeoipUnclaimed() + await pool.query("DELETE FROM mcp_drift WHERE id=$1",[geoipHold]) + await tickMcpWorkers();await tickMcpWorkers() + const {rows:[geoipTombstone]}=await pool.query(`SELECT t.action_id,t.target_global,t.target_server_ids::text[] AS server_ids, + NOT EXISTS(SELECT 1 FROM mcp_operations WHERE id=$1) AS operation_gone, + NOT EXISTS(SELECT 1 FROM mcp_plans WHERE id=$2) AS plan_gone, + NOT EXISTS(SELECT 1 FROM mcp_operation_steps WHERE operation_id=$1) AS step_gone, + NOT EXISTS(SELECT 1 FROM mcp_operation_checkpoints WHERE operation_id=$1) AS checkpoint_gone + FROM mcp_operation_replay_tombstones t WHERE t.operation_id=$1 AND t.plan_id=$2`,[geoip.id,geoip.planId]) + assert.deepEqual(geoipTombstone,{action_id:"geoip.settings.update",target_global:true,server_ids:[String(serverId)], + operation_gone:true,plan_gone:true,step_gone:true,checkpoint_gone:true}) + await assert.rejects(applyPlan(principal,geoip.planId,geoip.key),errorIs("OPERATION_HISTORY_EXPIRED",410)) + await assert.rejects(applyPlan(principal,randomUUID(),geoip.key),errorIs("STATE_CONFLICT",409)) + await assert.rejects(invokeMcpTool(foreign,"mm_operation_get",{operationId:geoip.id}),errorIs("NOT_FOUND",404)) + assert.deepEqual(await geoipState(),geoipBefore) + await pool.query("UPDATE backup_schedule_settings SET enabled=false,server_ids_json='[]' WHERE id=1") + await touchBackupScheduleRunMeta({lastRunAt:new Date().toISOString(),lastDurationMs:123,lastError:"owned backup scheduler metadata"}) + await pool.query(`INSERT INTO backup_entries(id,server_id,server_name,filename,size_bytes,kind,notes,created_at) + VALUES($1,$2,'owned fixture','owned-retention.rsc',123,'auto','keep',clock_timestamp())`,[randomUUID(),serverId]) + const backupState=async()=>(await pool.query(`SELECT to_jsonb(s) AS schedule, + (SELECT jsonb_agg(to_jsonb(b) ORDER BY id) FROM backup_entries b) AS entries + FROM backup_schedule_settings s WHERE id=1`)).rows[0] + const backupOutside=await operation(0,age+120000,"mm_backup_schedule_update_plan",{enabled:false,serverIds:[String(serverId+1000)]}) + const outsideState=await backupState() + await tickMcpWorkers();await tickMcpWorkers() + assert.equal((await pool.query("SELECT 1 FROM mcp_ordinary_retirement_claims WHERE operation_id=$1",[backupOutside.id])).rows.length,0) + assert.equal((await pool.query("SELECT 1 FROM mcp_operations WHERE id=$1",[backupOutside.id])).rows.length,1) + assert.deepEqual(await backupState(),outsideState) + await pool.query("UPDATE backup_schedule_settings SET server_ids_json='[]' WHERE id=1") + const backup=await operation(0,age+60000,"mm_backup_schedule_update_plan",{enabled:false,keepCount:9,serverIds:[String(serverId)]}) + const backupBefore=await backupState() + assert.equal(backupBefore.schedule.enabled,false);assert.equal(backupBefore.schedule.keep_count,9) + assert.deepEqual(backupBefore.schedule.server_ids_json,[String(serverId)]) + assert.equal(backupBefore.schedule.last_duration_ms,123);assert.equal(backupBefore.schedule.last_error,"owned backup scheduler metadata") + assert.deepEqual(backupBefore.entries,outsideState.entries) + const {rows:[backupStep]}=await pool.query("SELECT result FROM mcp_operation_steps WHERE operation_id=$1",[backup.id]) + await pool.query("UPDATE mcp_operation_steps SET result=jsonb_set(result,'{keepCount}','10') WHERE operation_id=$1",[backup.id]) + await tickMcpWorkers();await tickMcpWorkers() + assert.equal((await pool.query("SELECT 1 FROM mcp_ordinary_retirement_claims WHERE operation_id=$1",[backup.id])).rows.length,0) + assert.deepEqual(await backupState(),backupBefore) + await pool.query("UPDATE mcp_operation_steps SET result=$2 WHERE operation_id=$1",[backup.id,backupStep.result]) + await tickMcpWorkers();await tickMcpWorkers() + const {rows:[backupTombstone]}=await pool.query(`SELECT t.action_id,t.target_global,t.target_server_ids::text[] AS server_ids, + NOT EXISTS(SELECT 1 FROM mcp_operations WHERE id=$1) AS operation_gone, + NOT EXISTS(SELECT 1 FROM mcp_plans WHERE id=$2) AS plan_gone, + NOT EXISTS(SELECT 1 FROM mcp_operation_steps WHERE operation_id=$1) AS step_gone, + NOT EXISTS(SELECT 1 FROM mcp_operation_checkpoints WHERE operation_id=$1) AS checkpoint_gone + FROM mcp_operation_replay_tombstones t WHERE t.operation_id=$1 AND t.plan_id=$2`,[backup.id,backup.planId]) + assert.deepEqual(backupTombstone,{action_id:"backups.schedule.update",target_global:true,server_ids:[String(serverId)], + operation_gone:true,plan_gone:true,step_gone:true,checkpoint_gone:true}) + await assert.rejects(applyPlan(principal,backup.planId,backup.key),errorIs("OPERATION_HISTORY_EXPIRED",410)) + await assert.rejects(applyPlan(principal,randomUUID(),backup.key),errorIs("STATE_CONFLICT",409)) + await assert.rejects(invokeMcpTool(foreign,"mm_operation_get",{operationId:backup.id}),errorIs("NOT_FOUND",404)) + assert.deepEqual(await backupState(),backupBefore) await pool.query("UPDATE mcp_connections SET policy=jsonb_set(policy,'{scope,serverIds}','[]') WHERE id=$1",[principal.connectionId]) await assert.rejects(applyPlan(principal,due.planId,due.key),errorIs("NOT_FOUND",404)) for(const {record} of telemetry)await assert.rejects(applyPlan(principal,record.planId,record.key),errorIs("NOT_FOUND",404)) - console.log("Owned settings operations: five exact families, authenticated apply, physical 180-day purge, preserved settings and scoped replay") + await assert.rejects(applyPlan(principal,geoip.planId,geoip.key),errorIs("NOT_FOUND",404)) + await assert.rejects(applyPlan(principal,backup.planId,backup.key),errorIs("NOT_FOUND",404)) + console.log("Owned settings operations: seven exact families, authenticated apply, physical 180-day purge, preserved settings and scoped replay") }finally{resumeSchedulerJobs();await closePool()} diff --git a/backend/src/modules/mcp/ordinary-operation-retirement.ts b/backend/src/modules/mcp/ordinary-operation-retirement.ts index 4bfaf74e..eb8a7b29 100644 --- a/backend/src/modules/mcp/ordinary-operation-retirement.ts +++ b/backend/src/modules/mcp/ordinary-operation-retirement.ts @@ -1,5 +1,7 @@ import { randomUUID } from "node:crypto" import { logSettingsPatchSchema, logSettingsSchema } from "@mmapp/contracts" +import { geoipSettingsDtoSchema, geoipSettingsPatchSchema } from "@mmapp/contracts/geoip" +import { backupScheduleSettingsDtoSchema, putBackupScheduleSettingsSchema } from "@mmapp/contracts/backups" import { z } from "zod" import { connectProbeStorage } from "../../services/server-probe-storage-connection.js" import type { ProbeStorageSql } from "../../services/server-probe-journal-maintenance.js" @@ -15,6 +17,10 @@ const ids = z.array(positive).max(1000) const retention = z.number().int().min(1).max(365) const bool = z.boolean() const interval = (min:number,max:number)=>z.number().int().min(min).max(max) +const backupServerIds=z.array(z.string().regex(/^[1-9]\d{0,15}$/).refine(value=>Number.isSafeInteger(Number(value)))) + .max(1000).refine(values=>new Set(values).size===values.length) +const backupSchedule=backupScheduleSettingsDtoSchema.extend({serverIds:backupServerIds}).strict() +const backupBefore=z.object({value:backupSchedule,serverIds:ids}).strict() // Drizzle's string-mode timestamptz keeps PostgreSQL's space and hour-only // offset. Accept that exact wire form as well as an ISO timestamptz. const timestamp = z.string().regex(/^\d{4}-\d\d-\d\d[ T]\d\d:\d\d:\d\d(?:\.\d{1,6})?(?:Z|[+-]\d\d(?::\d\d)?)$/) @@ -37,6 +43,18 @@ const families={ speedEnabled:bool,intervalSec:interval(5,86400),probeIntervalSec:interval(5,86400),speedIntervalSec:interval(10,86400),retentionDays:retention}),true,true), "servers.api-ping.settings.update":telemetryFamily(2,"servers-api-ping-settings",z.object({enabled:bool,intervalSec:interval(10,86400)}),false,true), "internet-path.settings.update":telemetryFamily(1,"internet-path-settings",z.object({enabled:bool,intervalSec:interval(30,86400),retentionDays:retention}),true,false), + "geoip.settings.update":{ + version:1,domain:"geoip-settings", + patch:geoipSettingsPatchSchema.strict().refine(value=>Object.keys(value).length>0), + before:z.object({value:z.object({enabled:bool,updateIntervalSec:positive}).strict(),serverIds:ids}).strict(), + effect:z.object({settings:geoipSettingsDtoSchema.strict(),verificationScope:z.literal("configuration-persistence")}) + .strict().transform(result=>result.settings), + }, + "backups.schedule.update":{ + version:1,domain:"backup-schedule", + patch:putBackupScheduleSettingsSchema.extend({serverIds:backupServerIds.optional()}).strict().refine(value=>Object.keys(value).length>0), + before:backupBefore,effect:backupSchedule, + }, } as const const planEvidence = z.object({ target:z.object({scope:z.literal("global")}).strict(), @@ -64,7 +82,7 @@ const ELIGIBLE = `FROM public.mcp_operations o JOIN public.mcp_operation_checkpoints k ON k.operation_id=o.id AND k.step_index=0 WHERE (p.operation_id,p.operation_version) IN ( ('logs.settings.update',1),('traffic.settings.update',2),('uptime.settings.update',2), - ('servers.api-ping.settings.update',2),('internet-path.settings.update',1) + ('servers.api-ping.settings.update',2),('internet-path.settings.update',1),('geoip.settings.update',1),('backups.schedule.update',1) ) AND p.state='completed' AND o.state='applied' AND o.verification='passed' AND o.error_code IS NULL AND isfinite(o.finished_at) AND o.finished_ateffect[key]===value)) + requireThat(Object.entries(patch).every(([key,value])=>fingerprint(effect[key])===fingerprint(value))) requireThat(fingerprint(operation.result)===fingerprint({confirmedWrites:1,verificationScope:"configuration"})) for(const value of [checkpoint.before_sealed,checkpoint.after_sealed]){ requireThat(typeof value==="string"&&value.length<=1024&&fingerprint(await open({sealed:value}))===fingerprint([])) @@ -145,6 +163,10 @@ async function prepare(sql:ProbeStorageSql, operationId:string, connectionId:str &&domainObjects.length===1&&domainObjects[0].kind===family.domain&&domainObjects[0].id==="1"&&domainObjects[0].serverId===undefined) requireThat("serverIds" in before&&fingerprint(before.serverIds)===fingerprint(serverIds)) } + if(action==="backups.schedule.update"){ + const scope=new Set(serverIds.map(String)) + requireThat([...backupBefore.parse(before).value.serverIds,...backupSchedule.parse(effect).serverIds].every(id=>scope.has(id))) + } const signer=await privateFingerprinterExisting("mcp-ordinary-retirement-source-v1") return {planId:uuid.parse(plan.id),action,serverIds,key:z.string().min(1).max(200).parse(operation.idempotency_key),proof:signer(fingerprint(raw))} } diff --git a/backend/src/modules/mcp/routing-model.test.ts b/backend/src/modules/mcp/routing-model.test.ts index 3654c6f2..b1e7aafb 100644 --- a/backend/src/modules/mcp/routing-model.test.ts +++ b/backend/src/modules/mcp/routing-model.test.ts @@ -43,14 +43,43 @@ test("new unknown attempts block older successful data independently for each me test("failover requires three distinct failures and a freshly verified eligible standby", () => { const ranking = rankRoutingCandidates(profile, [...a, sample("b1", 1, 4, "wan-b"), sample("b2", 4, 3, "exit-b")], now) let state: ControllerState = {} - const health = (id: string, status: PathHealth["status"] = "failed"): PathHealth[] => [{ id, candidateId: "a", status, observedAt: at }, { id: "reserve", candidateId: "b", status: "passed", observedAt: at }] + const health = (id: string, time = now, status: PathHealth["status"] = "failed"): PathHealth[] => { + const observedAt = new Date(time).toISOString() + return [{ id, candidateId: "a", status, observedAt }, { id: "reserve", candidateId: "b", status: "passed", observedAt }] + } for (let i = 0; i < 5; i++) { const next = decideRoutingSwitch(profile, state, ranking, "a", now, health("one")); state = next.state; assert.equal(next.switchTo, null) } assert.equal(state.failureCount, 1) - const second = decideRoutingSwitch(profile, state, ranking, "a", now, health("two")); assert.equal(second.switchTo, null) - assert.equal(decideRoutingSwitch(profile, second.state, ranking, "a", now, health("three")).switchTo, "b") - assert.equal(decideRoutingSwitch(profile, second.state, ranking, "a", now, health("three", "unknown")).switchTo, null) - assert.equal(decideRoutingSwitch(profile, { ...second.state, switches: [at, at, at] }, ranking, "a", now, health("three")).switchTo, null) - assert.equal(decideRoutingSwitch(profile, second.state, ranking, "a", now + 100000, health("three")).switchTo, null) + const secondAt = now + 60000, thirdAt = now + 120000 + const second = decideRoutingSwitch(profile, state, ranking, "a", secondAt, health("two", secondAt)); assert.equal(second.switchTo, null) + assert.equal(decideRoutingSwitch(profile, second.state, ranking, "a", thirdAt, health("three", thirdAt)).switchTo, "b") + const unknown = decideRoutingSwitch(profile, second.state, ranking, "a", thirdAt, health("three", thirdAt, "unknown")) + assert.equal(unknown.switchTo, null) + assert.equal(unknown.state.failureObservedAt, undefined) + assert.equal(decideRoutingSwitch(profile, { ...second.state, switches: [at, at, at] }, ranking, "a", thirdAt, health("three", thirdAt)).switchTo, null) + assert.equal(decideRoutingSwitch(profile, second.state, ranking, "a", thirdAt + 100000, health("three", thirdAt)).switchTo, null) +}) +test("failover starts a fresh series after downtime or legacy state and ignores reordered failures", () => { + const ranking = rankRoutingCandidates(profile, [...a, sample("b1", 1, 4, "wan-b"), sample("b2", 4, 3, "exit-b")], now) + const last = now + 60000 + const persisted: ControllerState = JSON.parse(JSON.stringify({ failureCount: 2, failureKey: "two", + failureObservedAt: new Date(last).toISOString(), failingCandidateId: "a" })) + const check = (state: ControllerState, id: string, time: number, clock = time) => { + const observedAt = new Date(time).toISOString() + return decideRoutingSwitch(profile, state, ranking, "a", clock, + [{ id, candidateId: "a", status: "failed", observedAt }, { id: "reserve", candidateId: "b", status: "passed", observedAt }]) + } + for (const result of [check(persisted, "two", last + 60000), check(persisted, "older", last - 1000, last), + check(persisted, "clock-rollback", last - 1000), check(persisted, "equal-time", last)]) { + assert.equal(result.switchTo, null) + assert.equal(result.state.failureCount, 2) + assert.equal(result.state.failureObservedAt, persisted.failureObservedAt) + } + for (const [state, time] of [[persisted, now + 4 * 3600000], [{ ...persisted, failureObservedAt: undefined }, last + 60000]] as const) { + const result = check(state, "next", time) + assert.equal(result.switchTo, null) + assert.equal(result.state.failureCount, 1) + assert.equal(result.state.failureObservedAt, new Date(time).toISOString()) + } }) test("routing combines RTT, independent losses and bottleneck speed, never claims an observed path", () => { const ranked = rankRoutingCandidates(profile, a, now).find(row => row.candidate.id === "a")! @@ -92,3 +121,63 @@ test("controller does not count polling the same sample as consecutive confirmat assert.equal(decideRoutingSwitch(profile, { ...state, switches: [at, at, at] }, ranks, "a", now).switchTo, null) assert.equal(decideRoutingSwitch(profile, state, ranks, null, now).switchTo, null) }) + +test("hysteresis retains pending confirmation without counting sub-threshold observations or releasing cooldown", () => { + const ranked = rankRoutingCandidates(profile, [...a, sample("b1", 1, 4, "wan-b"), sample("b2", 4, 3, "exit-b")], now) + const compare = (state: ControllerState, gain: number, observation: string, time = now) => { + const ranks = ["b", "a"].map(id => ({ ...ranked.find(row => row.candidate.id === id)!, + score: id === "a" ? 50 : 50 + gain, observationIds: [id + observation] })) + return decideRoutingSwitch(profile, state, ranks, "a", time) + } + + const initial = compare({}, 15, "1").state + assert.equal(initial.consecutive, 1) + for (const [gain, observation] of [[6, "2"], [5, "3"]] as const) { + const result = compare(initial, gain, observation) + assert.equal(result.switchTo, null) + assert.deepEqual(result.state, initial) + } + const reset = compare(initial, 4, "4") + assert.equal(reset.state.pendingCandidateId, undefined) + assert.equal(reset.state.consecutive, 0) + + const second = compare(initial, 15, "5") + assert.equal(second.switchTo, null) + const third = compare({ ...second.state, lastSwitchAt: at }, 15, "6") + assert.equal(third.state.consecutive, 3) + assert.equal(third.switchTo, null) + + const afterCooldown = now + profile.cooldownSec * 1000 + assert.equal(compare(third.state, 6, "7", afterCooldown).switchTo, null) + assert.equal(compare(third.state, 15, "8", afterCooldown).switchTo, "b") +}) + +test("a different pending candidate cannot borrow confirmations from a shared observation key", () => { + const ranks = rankRoutingCandidates(profile, [...a, sample("b1", 1, 4, "wan-b"), sample("b2", 4, 3, "exit-b")], now) + .map(row => ({ ...row, score: row.candidate.id === "a" ? 50 : 75 })) + .sort((left, right) => right.score - left.score) + const observationKey = ranks.flatMap(row => row.observationIds).sort().join(":") + const persisted: ControllerState = { pendingCandidateId: "previous-candidate", consecutive: 3, observationKey } + const selected = decideRoutingSwitch(profile, persisted, ranks, "a", now) + assert.equal(selected.switchTo, null) + assert.equal(selected.state.pendingCandidateId, "b") + assert.equal(selected.state.consecutive, 0) + assert.equal(selected.state.observationKey, observationKey) + + const replay = decideRoutingSwitch(profile, selected.state, ranks, "a", now) + assert.equal(replay.switchTo, null) + assert.equal(replay.state.consecutive, 0) + + const freshRanks = [{ ...ranks[0], observationIds: ["new-b-1", "new-b-2"] }, ranks[1]] + const fresh = decideRoutingSwitch(profile, replay.state, freshRanks, "a", now) + assert.equal(fresh.state.consecutive, 1) + assert.equal(fresh.switchTo, null) + + const belowFloor = [{ ...freshRanks[0], score: 54 }, freshRanks[1]] + const reset = decideRoutingSwitch(profile, fresh.state, belowFloor, "a", now) + assert.equal(reset.state.pendingCandidateId, undefined) + assert.equal(reset.state.consecutive, 0) + const replayAfterReset = decideRoutingSwitch(profile, reset.state, freshRanks, "a", now) + assert.equal(replayAfterReset.switchTo, null) + assert.equal(replayAfterReset.state.consecutive, 0) +}) diff --git a/backend/src/modules/mcp/routing-model.ts b/backend/src/modules/mcp/routing-model.ts index 7301cad6..a6dccb84 100644 --- a/backend/src/modules/mcp/routing-model.ts +++ b/backend/src/modules/mcp/routing-model.ts @@ -51,36 +51,47 @@ export function rankRoutingCandidates(profile: RouteAiProfile, observations: Rou } export interface PathHealth { id: string; candidateId: string; status: "passed" | "failed" | "unknown"; observedAt: string } -export interface ControllerState { pendingCandidateId?: string; consecutive?: number; observationKey?: string; lastSwitchAt?: string; switches?: string[]; operationId?: string; pausedReason?: string; failureCount?: number; failureKey?: string; failingCandidateId?: string } +export interface ControllerState { pendingCandidateId?: string; consecutive?: number; observationKey?: string; lastSwitchAt?: string; switches?: string[]; operationId?: string; pausedReason?: string; failureCount?: number; failureKey?: string; failureObservedAt?: string; failingCandidateId?: string } +const pathHealthFreshnessSec = 90 export function decideRoutingSwitch(profile: RouteAiProfile, state: ControllerState, ranking: ReturnType, currentCandidateId: string | null, now = Date.now(), health: PathHealth[] = [], chargedSwitchesPerHour?: number) { const next: ControllerState = { ...state, switches: (state.switches ?? []).filter(at => now - Date.parse(at) < 3600000) } if (chargedSwitchesPerHour !== undefined && (!Number.isSafeInteger(chargedSwitchesPerHour) || chargedSwitchesPerHour < next.switches!.length || chargedSwitchesPerHour > 4096)) throw Error("Invalid common routing charge count") const switchCount = chargedSwitchesPerHour ?? next.switches!.length const best = ranking.find(candidate => candidate.eligible), current = ranking.find(candidate => candidate.candidate.id === currentCandidateId) - const currentHealth = health.find(value => value.candidateId === currentCandidateId && fresh(value.observedAt, 90, now)) + const currentHealth = health.find(value => value.candidateId === currentCandidateId && fresh(value.observedAt, pathHealthFreshnessSec, now)) if (currentHealth?.status === "failed" && currentCandidateId && !state.pausedReason) { - if (state.failureKey !== currentHealth.id) { - next.failureCount = state.failingCandidateId === currentCandidateId ? (state.failureCount ?? 0) + 1 : 1 - next.failureKey = currentHealth.id; next.failingCandidateId = currentCandidateId + const observed = Date.parse(currentHealth.observedAt), previous = Date.parse(state.failureObservedAt ?? "") + const sameCandidate = state.failingCandidateId === currentCandidateId + if (state.failureKey === currentHealth.id) { + if (!sameCandidate || observed !== previous) return { state: next, switchTo: null, reason: "Ожидание новой проверки отказа" } + } else { + if (observed <= previous) return { state: next, switchTo: null, reason: "Ожидание новой проверки отказа" } + const continues = sameCandidate && fresh(state.failureObservedAt ?? null, pathHealthFreshnessSec, observed) + next.failureCount = continues ? (state.failureCount ?? 0) + 1 : 1 + next.failureKey = currentHealth.id; next.failureObservedAt = currentHealth.observedAt; next.failingCandidateId = currentCandidateId } - const reserve = ranking.find(value => value.eligible && value.candidate.id !== currentCandidateId && health.some(check => check.candidateId === value.candidate.id && check.status === "passed" && fresh(check.observedAt, 90, now))) + const reserve = ranking.find(value => value.eligible && value.candidate.id !== currentCandidateId && health.some(check => check.candidateId === value.candidate.id && check.status === "passed" && fresh(check.observedAt, pathHealthFreshnessSec, now))) if ((next.failureCount ?? 0) < 3 || !reserve) return { state: next, switchTo: null, reason: "Failover ожидает три отдельных отказа и подтверждённый резерв" } if (switchCount >= profile.maxSwitchesPerHour) return { state: { ...next, pausedReason: "Лимит переключений" }, switchTo: null, reason: "Исчерпан часовой бюджет, включая failover" } if (next.lastSwitchAt && now - Date.parse(next.lastSwitchAt) < profile.cooldownSec * 1000) return { state: next, switchTo: null, reason: "Failover ожидает окончания cooldown" } return { state: { ...next, failureCount: 0 }, switchTo: reserve.candidate.id, reason: "Failover: три подтверждённых отказа и проверенный резерв" } } - next.failureCount = 0; next.failureKey = undefined; next.failingCandidateId = undefined + next.failureCount = 0; next.failureKey = undefined; next.failureObservedAt = undefined; next.failingCandidateId = undefined if (state.pausedReason || !currentCandidateId || !best || best.candidate.id === currentCandidateId) return { state: { ...next, pendingCandidateId: undefined, consecutive: 0 }, switchTo: null, reason: "Нет подтверждённого улучшения или текущего пути" } if (!current?.eligible || current.score === null) return { state: next, switchTo: null, reason: "Состояние текущего пути неизвестно; failover требует отдельных подтверждённых проверок отказа" } const gain = best.score! - current.score if (gain < Math.max(0, profile.switchThreshold - profile.hysteresis)) return { state: { ...next, consecutive: 0, pendingCandidateId: undefined }, switchTo: null, reason: "Сработал гистерезис" } - if (gain < profile.switchThreshold && next.pendingCandidateId !== best.candidate.id) return { state: next, switchTo: null, reason: "Порог улучшения не достигнут" } + if (gain < profile.switchThreshold) return { state: next, switchTo: null, reason: "Порог улучшения не достигнут" } const observationKey = [...best.observationIds, ...current.observationIds].sort().join(":") + if (next.pendingCandidateId !== undefined && next.pendingCandidateId !== best.candidate.id) { + next.pendingCandidateId = best.candidate.id + next.consecutive = 0 + } if (observationKey !== next.observationKey) { next.consecutive = next.pendingCandidateId === best.candidate.id ? (next.consecutive ?? 0) + 1 : 1 next.pendingCandidateId = best.candidate.id; next.observationKey = observationKey } - if ((next.consecutive ?? 0) < profile.confirmationSamples) return { state: next, switchTo: null, reason: "Ожидание новых измерений" } + if (next.pendingCandidateId !== best.candidate.id || (next.consecutive ?? 0) < profile.confirmationSamples) return { state: next, switchTo: null, reason: "Ожидание новых измерений" } if (next.lastSwitchAt && now - Date.parse(next.lastSwitchAt) < profile.cooldownSec * 1000) return { state: next, switchTo: null, reason: "Cooldown" } if (switchCount >= profile.maxSwitchesPerHour) return { state: { ...next, pausedReason: "Лимит переключений" }, switchTo: null, reason: "Исчерпан часовой бюджет переключений" } return { state: next, switchTo: best.candidate.id, reason: "Улучшение подтверждено свежими наблюдениями" } diff --git a/backend/src/services/server-poll-terminal-archive.owned.test.ts b/backend/src/services/server-poll-terminal-archive.owned.test.ts index 5d5e57ec..641d4c20 100644 --- a/backend/src/services/server-poll-terminal-archive.owned.test.ts +++ b/backend/src/services/server-poll-terminal-archive.owned.test.ts @@ -77,7 +77,7 @@ async function main() { } try { await applySqlMigrations(pool) - assert.equal(MIGRATION_MANIFEST.at(-1)?.file, "0053_server_probe_terminal_proofs.sql") + assert.equal(MIGRATION_MANIFEST.at(-1)?.file, "0055_mcp_backup_schedule_retirement.sql") assert.equal(MIGRATION_MANIFEST.find(pin=>pin.file==="0041_server_poll_terminal_archives.sql")?.sha256,"fbb8935ce977cfc507801f01beb8f303d4817ed63c6bdd4c05799768dfaac9d4") assert.equal((await pool.query("SELECT count(*)::int AS n FROM schema_migrations")).rows[0].n, MIGRATION_MANIFEST.length) assert.equal((await pool.query("SELECT count(*)::int AS n FROM server_poll_storage_admissions")).rows[0].n, 0)