From 8ce72c12690f296379d6296052f69379b2dd4f42 Mon Sep 17 00:00:00 2001 From: sunlei Date: Sun, 26 Jul 2026 19:51:10 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=B6=88=E9=99=A4=20Keeper=20=E9=80=9A?= =?UTF-8?q?=E9=81=93=E5=88=87=E6=8D=A2=E7=AB=9E=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../network-port-forward-group.service.ts | 118 +++++++++++------- .../network-management.service.spec.ts | 58 ++++++++- 2 files changed, 124 insertions(+), 52 deletions(-) diff --git a/src/modules/admin/platform-config/network-management/network-port-forward-group.service.ts b/src/modules/admin/platform-config/network-management/network-port-forward-group.service.ts index 24d75a0..7b201ab 100644 --- a/src/modules/admin/platform-config/network-management/network-port-forward-group.service.ts +++ b/src/modules/admin/platform-config/network-management/network-port-forward-group.service.ts @@ -42,6 +42,11 @@ type GroupTransactionResult = { group: NetworkPortForwardGroup; }; +type V1ChannelMutationTarget = { + action: string; + channelId: string; +}; + @Injectable() export class NetworkPortForwardGroupService { constructor( @@ -264,8 +269,32 @@ export class NetworkPortForwardGroupService { } async enableKeeper(groupId: string) { + return this.enableKeeperTarget(groupId); + } + + async disableKeeper(groupId: string) { + return this.disableKeeperTarget(groupId); + } + + async enableKeeperV1(channelId: string) { + return this.enableKeeperTarget({ action: 'UDP Keeper', channelId }); + } + + async disableKeeperV1(channelId: string) { + return this.disableKeeperTarget({ action: 'UDP Keeper', channelId }); + } + + async probeV1(channelId: string) { + return this.probeTarget({ action: 'UDP Keeper', channelId }); + } + + async probe(groupId: string) { + return this.probeTarget(groupId); + } + + private enableKeeperTarget(target: string | V1ChannelMutationTarget) { return this.mutateChannel( - groupId, + target, 'udp', async (_, channel, channels) => { if (channel.keeperDesiredEnabled) return false; @@ -280,9 +309,9 @@ export class NetworkPortForwardGroupService { ); } - async disableKeeper(groupId: string) { + private disableKeeperTarget(target: string | V1ChannelMutationTarget) { return this.mutateChannel( - groupId, + target, 'udp', async (_, channel, channels) => { if (!channel.keeperDesiredEnabled) return false; @@ -298,27 +327,9 @@ export class NetworkPortForwardGroupService { ); } - async enableKeeperV1(channelId: string) { - return this.enableKeeper( - await this.resolveV1GroupId(channelId, 'udp', 'UDP Keeper'), - ); - } - - async disableKeeperV1(channelId: string) { - return this.disableKeeper( - await this.resolveV1GroupId(channelId, 'udp', 'UDP Keeper'), - ); - } - - async probeV1(channelId: string) { - return this.probe( - await this.resolveV1GroupId(channelId, 'udp', 'UDP Keeper'), - ); - } - - async probe(groupId: string) { + private probeTarget(target: string | V1ChannelMutationTarget) { return this.mutateChannel( - groupId, + target, 'udp', async (_, channel, channels) => { this.assertMechanismTransitionAllowed(channels); @@ -590,7 +601,7 @@ export class NetworkPortForwardGroupService { } private async mutateChannel( - groupId: string, + target: string | V1ChannelMutationTarget, protocol: PortForwardProtocol, change: ( group: NetworkPortForwardGroup, @@ -599,12 +610,46 @@ export class NetworkPortForwardGroupService { ) => Promise, invalidProtocolIsBadRequest = false, ) { - this.assertId(groupId, '逻辑组'); + const v1Target = typeof target === 'string' ? null : target; + const targetId = typeof target === 'string' ? target : target.channelId; + this.assertId(targetId, v1Target ? '端口转发' : '逻辑组'); const result = await this.dataSource.transaction(async (manager) => { const state = await this.lockAgentState(manager); + const mappingRepository = manager.getRepository(NetworkPortForward); + const requested = v1Target + ? await mappingRepository.findOne({ + lock: { mode: 'pessimistic_write' }, + where: { id: v1Target.channelId, isDeleted: false }, + }) + : null; + if (v1Target && !requested) { + throwVbenError('端口转发不存在', HttpStatus.NOT_FOUND); + } + if (v1Target && requested && requested.protocol !== protocol) { + throwVbenError( + `${requested.protocol.toUpperCase()} 通道不支持 ${v1Target.action}`, + HttpStatus.BAD_REQUEST, + ); + } + const groupId = requested ? String(requested.groupId) : targetId; const group = await this.findLockedGroup(manager, groupId); const channels = await this.findChannels(manager, groupId); - const channel = channels.find((item) => item.protocol === protocol); + const requestedIndex = requested + ? channels.findIndex((item) => item.id === requested.id) + : -1; + if ( + requested && + (requested.isDeleted || + requested.groupId !== group.id || + requestedIndex < 0 || + channels[requestedIndex].groupId !== group.id || + channels[requestedIndex].protocol !== protocol) + ) { + throwVbenError('端口转发通道已失效或被替换', HttpStatus.CONFLICT); + } + if (requested) channels[requestedIndex] = requested; + const channel = + requested || channels.find((item) => item.protocol === protocol); if (!channel) { throwVbenError( `逻辑组不包含 ${protocol.toUpperCase()} 协议通道`, @@ -620,7 +665,7 @@ export class NetworkPortForwardGroupService { if (!changed) return { changed, channel }; const revision = this.advanceGlobalRevision(state); this.assignRevision([channel], revision, state.desiredIssuedAt); - await manager.getRepository(NetworkPortForward).save(channel); + await mappingRepository.save(channel); await manager.getRepository(NetworkAgentState).save(state); return { changed, channel }; }); @@ -640,25 +685,6 @@ export class NetworkPortForwardGroupService { return group; } - private async resolveV1GroupId( - channelId: string, - protocol: PortForwardProtocol, - action: string, - ): Promise { - this.assertId(channelId, '端口转发'); - const channel = await this.mappingRepository.findOne({ - where: { id: channelId, isDeleted: false }, - }); - if (!channel) throwVbenError('端口转发不存在', HttpStatus.NOT_FOUND); - if (channel.protocol !== protocol) { - throwVbenError( - `${channel.protocol.toUpperCase()} 通道不支持 ${action}`, - HttpStatus.BAD_REQUEST, - ); - } - return String(channel.groupId); - } - private async findChannels( manager: EntityManager, groupId: string, diff --git a/test/admin/network-management/network-management.service.spec.ts b/test/admin/network-management/network-management.service.spec.ts index 7b2f5fe..5b3312a 100644 --- a/test/admin/network-management/network-management.service.spec.ts +++ b/test/admin/network-management/network-management.service.spec.ts @@ -16,13 +16,16 @@ type Harness = { bootstrapExecute: jest.Mock; groups: NetworkPortForwardGroup[]; histories: NetworkEndpointHistory[]; + injectedMappingFindOne: jest.Mock; mappings: NetworkPortForward[]; mqtt: jest.Mocked>; service: NetworkManagementService; state: NetworkAgentState; + transactionalMappingFindOne: jest.Mock; }; type ReleaseConfig = { + beforeTransaction?: () => void; canaryPorts?: string; mode?: string; }; @@ -56,6 +59,12 @@ function createHarness( bootstrapOrder.push('insert-ignore'); return { identifiers: [] }; }); + const findMapping = async ({ where }: { where: Record }) => + mappings.find((mapping) => + Object.entries(where).every(([key, value]) => mapping[key] === value), + ) || null; + const injectedMappingFindOne = jest.fn(findMapping); + const transactionalMappingFindOne = jest.fn(findMapping); const state = Object.assign(new NetworkAgentState(), { agentId: 'nas-main', appliedRevision: '0', @@ -74,10 +83,7 @@ function createHarness( mappings.filter((mapping) => Object.entries(where).every(([key, value]) => mapping[key] === value), ), - findOne: async ({ where }) => - mappings.find((mapping) => - Object.entries(where).every(([key, value]) => mapping[key] === value), - ) || null, + findOne: injectedMappingFindOne, save: async (value) => { const values = Array.isArray(value) ? value : [value]; for (const mapping of values) { @@ -88,6 +94,10 @@ function createHarness( return value; }, } as unknown as Repository; + const transactionalMappingRepository = { + ...mappingRepository, + findOne: transactionalMappingFindOne, + } as unknown as Repository; const groupRepository = { create: (input) => Object.assign(new NetworkPortForwardGroup(), { id: '200' }, input), @@ -128,7 +138,7 @@ function createHarness( } as unknown as Repository; const manager = { getRepository: (entity) => { - if (entity === NetworkPortForward) return mappingRepository; + if (entity === NetworkPortForward) return transactionalMappingRepository; if (entity === NetworkPortForwardGroup) return groupRepository; if (entity === NetworkAgentState) return stateRepository; if (entity === NetworkEndpointHistory) return historyRepository; @@ -136,7 +146,10 @@ function createHarness( }, } as unknown as EntityManager; const dataSource = { - transaction: async (work) => await work(manager), + transaction: async (work) => { + releaseConfig.beforeTransaction?.(); + return await work(manager); + }, } as unknown as DataSource; const configService = { get: (key) => @@ -173,10 +186,12 @@ function createHarness( bootstrapOrder, groups, histories, + injectedMappingFindOne, mappings, mqtt, service, state, + transactionalMappingFindOne, }; } @@ -674,4 +689,35 @@ describe('NetworkManagementService', () => { status: 404, }); }); + + it('locks the exact v1 channel ID in-transaction and never mutates its replacement', async () => { + const stale = createMapping({ id: '100' }); + const replacement = createMapping({ id: '102', probeRequestId: null }); + const mappings = [stale]; + const harness = createHarness(mappings, { + beforeTransaction: () => { + stale.activeGroupProtocolKey = null; + stale.activeKey = null; + stale.isDeleted = true; + mappings.push(replacement); + }, + mode: 'off', + }); + + await expect(harness.service.enableKeeper('100')).rejects.toMatchObject({ + status: 404, + }); + expect(harness.transactionalMappingFindOne).toHaveBeenCalledWith({ + lock: { mode: 'pessimistic_write' }, + where: { id: '100', isDeleted: false }, + }); + expect(harness.injectedMappingFindOne).not.toHaveBeenCalled(); + expect(replacement).toMatchObject({ + desiredRevision: '3', + keeperDesiredEnabled: false, + probeRequestId: null, + }); + expect(harness.state.desiredRevision).toBe('3'); + expect(harness.mqtt.requestDesiredPublish).not.toHaveBeenCalled(); + }); });