General Actions to handlers Refacto
This commit is contained in:
STEINNI
@@ -0,0 +1,17 @@
|
||||
import { assembleHandlers, createDispatchMessage } from '../../../bus/assembleMesh.js'
|
||||
import * as prepare from './prepare.js'
|
||||
|
||||
const { actionHandlers, eventHandlers, afterLogin } = assembleHandlers([prepare])
|
||||
|
||||
export { actionHandlers, afterLogin }
|
||||
|
||||
export const dispatchMessage = createDispatchMessage({
|
||||
eventHandlers,
|
||||
actionRules(redisCnx) {
|
||||
const maestro = redisCnx.config.maestro ?? {}
|
||||
const arenaChannel = maestro.bus?.arena?.actionsChannel
|
||||
return({
|
||||
channels: arenaChannel ? [arenaChannel] : [],
|
||||
})
|
||||
},
|
||||
})
|
||||
@@ -0,0 +1,9 @@
|
||||
|
||||
export const eventHandlers = {
|
||||
'arena:gods:ready': {
|
||||
readyToStart(msg, chan) {
|
||||
if(!this.maestroSrv) return
|
||||
this.maestroSrv.handlePrepareAck(msg, chan)
|
||||
},
|
||||
},
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
import { assembleHandlers, createDispatchMessage } from '../../../bus/assembleMesh.js'
|
||||
import * as simulation from './simulation.js'
|
||||
import * as utilities from './utilities.js'
|
||||
|
||||
const { actionHandlers, eventHandlers, afterLogin } = assembleHandlers([simulation, utilities])
|
||||
|
||||
export { actionHandlers, afterLogin }
|
||||
|
||||
export const dispatchMessage = createDispatchMessage({
|
||||
eventHandlers,
|
||||
actionRules(redisCnx) {
|
||||
const maestro = redisCnx.config.maestro ?? {}
|
||||
return({
|
||||
channels: [maestro.maestroActionsChannel].filter(Boolean),
|
||||
})
|
||||
},
|
||||
})
|
||||
@@ -0,0 +1,63 @@
|
||||
import { replyToAction } from '../../../bus/publishActionReply.js'
|
||||
import { isValidUuid } from '../../simRepository.js'
|
||||
|
||||
export const actions = {
|
||||
|
||||
async action_STARTSIMULATION(action, payload, reqid, sender, roles) {
|
||||
if(!isValidUuid(sender)) {
|
||||
replyToAction(this, { action, reqid, sender, success: false, err: 'Missing or invalid sender (user UUID)' })
|
||||
return
|
||||
}
|
||||
|
||||
if(!payload?.simulationUuid) {
|
||||
replyToAction(this, { action, reqid, sender, success: false, err: 'Missing simulationUuid' })
|
||||
return
|
||||
}
|
||||
|
||||
const result = await this.maestroSrv.startSimulation(sender, payload)
|
||||
if(!result.ok) {
|
||||
replyToAction(this, { action, reqid, sender, success: false, err: result.err })
|
||||
return
|
||||
}
|
||||
|
||||
replyToAction(this, {
|
||||
action,
|
||||
reqid,
|
||||
sender,
|
||||
success: true,
|
||||
payload: {
|
||||
simulationId: result.simulationId,
|
||||
keyframeId: result.keyframeId,
|
||||
infraId: result.infraId,
|
||||
agentIds: result.agentIds,
|
||||
},
|
||||
})
|
||||
},
|
||||
|
||||
async action_STOPSIMULATION(action, payload, reqid, sender, roles) {
|
||||
if(!isValidUuid(sender)) {
|
||||
replyToAction(this, { action, reqid, sender, success: false, err: 'Missing or invalid sender (user UUID)' })
|
||||
return
|
||||
}
|
||||
|
||||
if(!payload?.simulationUuid) {
|
||||
replyToAction(this, { action, reqid, sender, success: false, err: 'Missing simulationUuid' })
|
||||
return
|
||||
}
|
||||
|
||||
const result = await this.maestroSrv.stopSimulation(sender, payload)
|
||||
if(!result.ok) {
|
||||
replyToAction(this, { action, reqid, sender, success: false, err: result.err })
|
||||
return
|
||||
}
|
||||
|
||||
replyToAction(this, {
|
||||
action,
|
||||
reqid,
|
||||
sender,
|
||||
success: true,
|
||||
payload: { simulationId: result.simulationId },
|
||||
})
|
||||
},
|
||||
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
import { replyToAction } from '../../../bus/publishActionReply.js'
|
||||
|
||||
export const actions = {
|
||||
|
||||
async action_RELOADCONFIG(action, payload, reqid, sender, roles) {
|
||||
this.reloadAccessRights()
|
||||
replyToAction(this, { action, reqid, sender, success: true })
|
||||
},
|
||||
|
||||
async action_GETCONFIG(action, payload, reqid, sender, roles) {
|
||||
replyToAction(this, {
|
||||
action,
|
||||
reqid,
|
||||
sender,
|
||||
success: true,
|
||||
payload: this.getAccessRights(),
|
||||
})
|
||||
},
|
||||
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
const RESERVED_HASH_FIELDS = new Set(['position', 'vector', 'speed', 'segment'])
|
||||
|
||||
export class ArenaGroom {
|
||||
|
||||
constructor(arenaCnx, arenaStorage, debug = false) {
|
||||
this.cnx = arenaCnx
|
||||
this.arenaStorage = arenaStorage
|
||||
this.debug = debug
|
||||
}
|
||||
|
||||
agentHashKey(agentId) {
|
||||
return(this.arenaStorage.agentHashKey.replace(/\[UID\]/g, agentId))
|
||||
}
|
||||
|
||||
async clearArena() {
|
||||
const ids = await this.cnx.redisSmembers(this.arenaStorage.agentsIndexKey)
|
||||
for(const id of ids) {
|
||||
await this.cnx.redisDel(this.agentHashKey(id))
|
||||
}
|
||||
await this.cnx.redisDel(this.arenaStorage.agentsIndexKey)
|
||||
if(this.debug) console.log(`[Maestro] Cleared arena store (${ids.length} agent(s))`)
|
||||
}
|
||||
|
||||
async seedAgents(agents) {
|
||||
for(const agent of agents) {
|
||||
const key = this.agentHashKey(agent.id)
|
||||
await this.cnx.redisHset(key, 'position', agent.position)
|
||||
await this.cnx.redisHset(key, 'vector', agent.vector)
|
||||
for(const [field, value] of Object.entries(agent.store ?? {})) {
|
||||
if(RESERVED_HASH_FIELDS.has(field)) continue
|
||||
await this.cnx.redisHset(key, field, value)
|
||||
}
|
||||
await this.cnx.redisSadd(this.arenaStorage.agentsIndexKey, agent.id)
|
||||
}
|
||||
if(this.debug) console.log(`[Maestro] Groomed ${agents.length} agent(s) into arena store`)
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,223 @@
|
||||
import { AccesRights } from '../accesRights.js'
|
||||
import { MySQLClient } from '@p42/p42modules'
|
||||
import { SimRepository } from './simRepository.js'
|
||||
import { ArenaGroom } from './arenaGroom.js'
|
||||
import { MaestroState } from './orchestrationState.js'
|
||||
import { PrepareQuorum } from './prepareQuorum.js'
|
||||
import { buildPrepareQuorum } from './primordialDaemons.js'
|
||||
|
||||
export class maestroServer {
|
||||
|
||||
constructor(configHelper, allRediscnx, debug) {
|
||||
this.configHelper = configHelper
|
||||
this.maestroConfig = configHelper.config
|
||||
this.allRediscnx = allRediscnx
|
||||
this.debug = debug
|
||||
this.accessRights = new AccesRights(this.maestroConfig, debug)
|
||||
this.arenaCnx = null
|
||||
this.arenaCnxs = []
|
||||
this.systemCnx = null
|
||||
this.db = null
|
||||
this.simRepo = null
|
||||
this.arenaGroom = null
|
||||
this.prepareQuorum = null
|
||||
this.orchestrationState = MaestroState.IDLE
|
||||
this.simulationId = null
|
||||
this.agentIds = []
|
||||
}
|
||||
|
||||
getMaestroSettings() {
|
||||
const maestro = this.maestroConfig.maestro ?? {}
|
||||
return({
|
||||
senderId: maestro.senderId ?? 'maestro',
|
||||
lifecycle: {
|
||||
arenaChannel: maestro.lifecycle?.arenaChannel ?? 'arena:lifecycle',
|
||||
prepareAckChannel: maestro.lifecycle?.godsReadyChannel ?? 'arena:gods:ready',
|
||||
},
|
||||
readyTimeoutMs: maestro.readyTimeoutMs ?? 30000,
|
||||
})
|
||||
}
|
||||
|
||||
getArenaStorageSettings() {
|
||||
const gps = this.maestroConfig.gps ?? {}
|
||||
return({
|
||||
agentHashKey: gps.arenaStorage?.agentHashKey ?? 'arena:agents:[UID]',
|
||||
agentsIndexKey: gps.arenaStorage?.agentsIndexKey ?? 'arena:agents',
|
||||
})
|
||||
}
|
||||
|
||||
isIdle() {
|
||||
return(this.orchestrationState === MaestroState.IDLE)
|
||||
}
|
||||
|
||||
isLive() {
|
||||
return(this.orchestrationState === MaestroState.LIVE)
|
||||
}
|
||||
|
||||
async init() {
|
||||
const mysqlCfg = this.maestroConfig.mysql
|
||||
if(!mysqlCfg) {
|
||||
console.error('[Maestro] Missing mysql config')
|
||||
return(false)
|
||||
}
|
||||
this.db = await MySQLClient.createPool(mysqlCfg)
|
||||
this.simRepo = new SimRepository(this.db, MySQLClient.resolveDatabases(mysqlCfg), this.debug)
|
||||
if(this.debug) console.log('[Maestro] MySQL pool ready')
|
||||
return(true)
|
||||
}
|
||||
|
||||
refreshArenaGroom() {
|
||||
if(this.arenaCnx) {
|
||||
this.arenaGroom = new ArenaGroom(
|
||||
this.arenaCnx,
|
||||
this.getArenaStorageSettings(),
|
||||
this.debug
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
refreshPrepareQuorum() {
|
||||
if(!this.arenaCnx) return
|
||||
const { prepareAckChannel, readyTimeoutMs } = this.getMaestroSettings()
|
||||
this.prepareQuorum = new PrepareQuorum({
|
||||
ackChannel: prepareAckChannel,
|
||||
timeoutMs: readyTimeoutMs,
|
||||
matchesChan: this.arenaCnx.matchesChan.bind(this.arenaCnx),
|
||||
debug: this.debug,
|
||||
})
|
||||
}
|
||||
|
||||
wireSystemConnexion(cnx) {
|
||||
cnx.maestroSrv = this
|
||||
cnx.accessRights = this.accessRights
|
||||
cnx.reloadAccessRights = () => this.reloadAccessRights()
|
||||
cnx.getAccessRights = () => this.getAccessRights()
|
||||
if(!this.systemCnx || cnx.redisConfig.role === 'primary') {
|
||||
this.systemCnx = cnx
|
||||
}
|
||||
}
|
||||
|
||||
wireArenaConnexion(cnx) {
|
||||
cnx.maestroSrv = this
|
||||
this.arenaCnxs.push(cnx)
|
||||
if(!this.arenaCnx || cnx.redisConfig.role === 'primary') {
|
||||
this.arenaCnx = cnx
|
||||
this.refreshArenaGroom()
|
||||
this.refreshPrepareQuorum()
|
||||
}
|
||||
}
|
||||
|
||||
resetOrchestration() {
|
||||
this.orchestrationState = MaestroState.IDLE
|
||||
this.simulationId = null
|
||||
this.agentIds = []
|
||||
this.prepareQuorum?.cancel()
|
||||
}
|
||||
|
||||
handlePrepareAck(msg, chan) {
|
||||
if(this.orchestrationState !== MaestroState.PREPARING) return(false)
|
||||
if(!this.prepareQuorum) return(false)
|
||||
return(this.prepareQuorum.handleMessage(msg, chan))
|
||||
}
|
||||
|
||||
async publishLifecycle(eventType, payload) {
|
||||
if(!this.arenaCnx) throw(new Error('No arena Redis connection'))
|
||||
const { arenaChannel, senderId } = this.getMaestroSettings().lifecycle
|
||||
await this.arenaCnx.redisPublish(arenaChannel, {
|
||||
eventType,
|
||||
sender: senderId,
|
||||
payload,
|
||||
})
|
||||
if(this.debug) console.log(`[Maestro] Published ${eventType} simulationId=${payload.simulationId}`)
|
||||
}
|
||||
|
||||
async startSimulation(userUuid, payload) {
|
||||
if(!this.simRepo) return({ ok: false, err: 'Database not initialized' })
|
||||
if(!this.arenaGroom) return({ ok: false, err: 'No arena Redis connection' })
|
||||
if(!this.prepareQuorum) return({ ok: false, err: 'No prepare quorum (arena Redis not wired)' })
|
||||
if(!this.isIdle()) return({ ok: false, err: 'A simulation is already in progress' })
|
||||
|
||||
const simulationUuid = payload?.simulationUuid
|
||||
const infraId = payload?.infraId ?? null
|
||||
|
||||
const access = await this.simRepo.validateSimulationAccess(userUuid, simulationUuid)
|
||||
if(!access.ok) return(access)
|
||||
|
||||
const keyframeId = access.sim.sim_root_kf_uuid
|
||||
const agentsResult = await this.simRepo.loadKeyframeAgents(keyframeId)
|
||||
if(!agentsResult.ok) return(agentsResult)
|
||||
|
||||
await this.arenaGroom.clearArena()
|
||||
await this.arenaGroom.seedAgents(agentsResult.agents)
|
||||
|
||||
this.simulationId = simulationUuid
|
||||
this.agentIds = agentsResult.agents.map(a => a.id)
|
||||
this.orchestrationState = MaestroState.PREPARING
|
||||
|
||||
const lifecyclePayload = {
|
||||
simulationId: this.simulationId,
|
||||
agentIds: this.agentIds,
|
||||
keyframeId,
|
||||
infraId,
|
||||
}
|
||||
|
||||
const expectedParticipants = buildPrepareQuorum(this.agentIds, this.maestroConfig)
|
||||
const readyWait = this.prepareQuorum.begin(expectedParticipants, this.simulationId)
|
||||
|
||||
await this.publishLifecycle('onYourMarks', lifecyclePayload)
|
||||
|
||||
const quorum = await readyWait
|
||||
if(!quorum.ok) {
|
||||
await this.arenaGroom.clearArena()
|
||||
this.resetOrchestration()
|
||||
return(quorum)
|
||||
}
|
||||
|
||||
await this.publishLifecycle('bigBang', {
|
||||
simulationId: this.simulationId,
|
||||
agentIds: this.agentIds,
|
||||
})
|
||||
|
||||
this.orchestrationState = MaestroState.LIVE
|
||||
if(this.debug) console.log(`[Maestro] LIVE simulationId=${this.simulationId}`)
|
||||
|
||||
return({
|
||||
ok: true,
|
||||
simulationId: this.simulationId,
|
||||
keyframeId,
|
||||
infraId,
|
||||
agentIds: this.agentIds,
|
||||
})
|
||||
}
|
||||
|
||||
async stopSimulation(userUuid, payload) {
|
||||
if(!this.simRepo) return({ ok: false, err: 'Database not initialized' })
|
||||
if(!this.arenaGroom) return({ ok: false, err: 'No arena Redis connection' })
|
||||
|
||||
const simulationUuid = payload?.simulationUuid
|
||||
const access = await this.simRepo.validateSimulationOwner(userUuid, simulationUuid)
|
||||
if(!access.ok) return(access)
|
||||
|
||||
if(this.simulationId && this.simulationId !== simulationUuid) {
|
||||
return({ ok: false, err: 'Another simulation is active' })
|
||||
}
|
||||
|
||||
await this.arenaGroom.clearArena()
|
||||
this.resetOrchestration()
|
||||
|
||||
if(this.debug) console.log(`[Maestro] Stopped simulationId=${simulationUuid}`)
|
||||
|
||||
return({ ok: true, simulationId: simulationUuid })
|
||||
}
|
||||
|
||||
async reloadAccessRights() {
|
||||
await this.configHelper.refreshAccessRights()
|
||||
this.maestroConfig.accessRights = this.configHelper.config.accessRights
|
||||
this.accessRights.refreshAccessRights(this.maestroConfig)
|
||||
}
|
||||
|
||||
getAccessRights() {
|
||||
return(this.maestroConfig.accessRights)
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
export const MaestroState = {
|
||||
IDLE: 'idle',
|
||||
PREPARING: 'preparing',
|
||||
LIVE: 'live',
|
||||
}
|
||||
@@ -0,0 +1,140 @@
|
||||
|
||||
import yargs from 'yargs/yargs'
|
||||
import { hideBin } from 'yargs/helpers'
|
||||
import 'node:process'
|
||||
import { RedisConnexion } from '../redisConnexion.js'
|
||||
import { busReplyRoute } from '../bus/publishActionReply.js'
|
||||
import { configHelper } from '../configHelper.js'
|
||||
import { maestroServer } from './maestroServer.js'
|
||||
import * as systemMesh from './actions/system/index.js'
|
||||
import * as arenaMesh from './actions/arena/index.js'
|
||||
|
||||
const meshModules = {
|
||||
system: systemMesh,
|
||||
arena: arenaMesh,
|
||||
}
|
||||
|
||||
const originalLog = console.log
|
||||
const originalWarn = console.warn
|
||||
const originalError = console.error
|
||||
function logWithTimestamp(originalFn, level, ...args) {
|
||||
const timestamp = new Date().toISOString()
|
||||
originalFn(`[${timestamp}] [${level}]`, ...args)
|
||||
}
|
||||
|
||||
console.log = (...args) => logWithTimestamp(originalLog, 'LOG', ...args)
|
||||
console.warn = (...args) => logWithTimestamp(originalWarn, 'WARN', ...args)
|
||||
console.error = (...args) => logWithTimestamp(originalError, 'ERROR', ...args)
|
||||
|
||||
const argv = yargs(hideBin(process.argv)).command('Maestro', 'Simulation orchestrator for P42', {})
|
||||
.options({
|
||||
'debug': {
|
||||
description: 'shows debug info',
|
||||
alias: 'd',
|
||||
defaut: false,
|
||||
type: 'boolean'
|
||||
},
|
||||
'config': {
|
||||
description: 'Points to config file (default: ../config.json)',
|
||||
alias: 'c',
|
||||
default: '../config.json',
|
||||
type: 'string'
|
||||
},
|
||||
}).help().version('1.0').argv
|
||||
|
||||
const debug = Boolean(process.env.DEBUG) || argv.debug
|
||||
|
||||
let cfgh = new configHelper({
|
||||
localfile: argv.config,
|
||||
})
|
||||
|
||||
function meshRedisConns(mesh, meshName, debug, rootConfig) {
|
||||
const { redis, ...meshConfig } = mesh
|
||||
const busRoute = busReplyRoute(rootConfig.maestro, meshName)
|
||||
return redis.map(cfg =>
|
||||
new RedisConnexion({
|
||||
debug,
|
||||
config: { ...cfg, ...meshConfig, maestro: rootConfig.maestro },
|
||||
redisId: cfg.redisId,
|
||||
meshName,
|
||||
meshModule: meshModules[meshName],
|
||||
senderId: busRoute?.senderId,
|
||||
actionsReply: busRoute?.actionsReply,
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
async function startAllRedis(rootConfig, cfgh) {
|
||||
if(debug) console.log('Starting all Redis instances...')
|
||||
|
||||
const redisConns = [
|
||||
...meshRedisConns(rootConfig.systemMesh, 'system', debug, rootConfig),
|
||||
...meshRedisConns(rootConfig.arenaMesh, 'arena', debug, rootConfig),
|
||||
]
|
||||
|
||||
const srv = new maestroServer(cfgh, redisConns, debug)
|
||||
for(const cnx of redisConns) {
|
||||
if(cnx.meshName === 'system') srv.wireSystemConnexion(cnx)
|
||||
else if(cnx.meshName === 'arena') srv.wireArenaConnexion(cnx)
|
||||
}
|
||||
|
||||
const loginResults = await Promise.allSettled(
|
||||
redisConns.map(async cnx => {
|
||||
await cnx.redisLogin()
|
||||
return(cnx.redisId)
|
||||
})
|
||||
)
|
||||
|
||||
const failedLogin = loginResults.filter(r => r.status !== 'fulfilled')
|
||||
if(failedLogin.length > 0) {
|
||||
console.error('Redis login failures:')
|
||||
failedLogin.forEach((r, i) => {
|
||||
const id = redisConns[i].redisId
|
||||
console.error(`login failed for redis:[${id}] → ${r.reason}`)
|
||||
})
|
||||
throw new Error(
|
||||
`Redis login failed for ${failedLogin.length}/${redisConns.length} instances`
|
||||
)
|
||||
}
|
||||
|
||||
if(debug) console.log('All Redis logins OK')
|
||||
|
||||
const chanResults = await Promise.allSettled(
|
||||
redisConns.map(async cnx => {
|
||||
await cnx.redisChansStart()
|
||||
return(cnx.redisId)
|
||||
})
|
||||
)
|
||||
|
||||
const failedChans = chanResults.filter(r => r.status !== 'fulfilled')
|
||||
if(failedChans.length > 0) {
|
||||
console.error('Redis chansStart failures:')
|
||||
failedChans.forEach((r, i) => {
|
||||
const id = redisConns[i].redisId
|
||||
console.error(`chansStart failed for redis:[${id}] → ${r.reason}`)
|
||||
})
|
||||
throw new Error(
|
||||
`Redis chansStart failed for ${failedChans.length}/${redisConns.length} instances`
|
||||
)
|
||||
}
|
||||
|
||||
if(debug) console.log('All Redis chansStart OK')
|
||||
|
||||
return({ redisConns, srv })
|
||||
}
|
||||
|
||||
cfgh.fetchConfig().then(async rootConfig => {
|
||||
if(!rootConfig) {
|
||||
console.error('Cannot get a valid configuration ! Aaarrghhh...')
|
||||
process.exit()
|
||||
}
|
||||
|
||||
console.log(`Debug mode : ${debug ? 'ON' : 'OFF'}`)
|
||||
|
||||
const { srv } = await startAllRedis(rootConfig, cfgh)
|
||||
const dbOk = await srv.init()
|
||||
if(!dbOk) {
|
||||
console.error('Maestro MySQL init failed — exiting')
|
||||
process.exit(1)
|
||||
}
|
||||
})
|
||||
@@ -0,0 +1,9 @@
|
||||
{
|
||||
"name": "p42Maestro",
|
||||
"version": "1.0.0",
|
||||
"description": "Simulation orchestrator God-daemon for P42",
|
||||
"type": "module",
|
||||
"dependencies": {
|
||||
"yargs": "^17.7.2"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
|
||||
export class PrepareQuorum {
|
||||
|
||||
constructor({ ackChannel, timeoutMs, matchesChan, debug = false }) {
|
||||
this.ackChannel = ackChannel
|
||||
this.timeoutMs = timeoutMs
|
||||
this.matchesChan = matchesChan
|
||||
this.debug = debug
|
||||
this.expected = new Set()
|
||||
this.ready = new Map()
|
||||
this.simulationId = null
|
||||
this.active = false
|
||||
this.resolve = null
|
||||
this.timer = null
|
||||
}
|
||||
|
||||
begin(expectedParticipantIds, simulationId) {
|
||||
this.cancel()
|
||||
this.expected = new Set(expectedParticipantIds)
|
||||
this.ready.clear()
|
||||
this.simulationId = simulationId
|
||||
this.active = true
|
||||
|
||||
if(this.debug) {
|
||||
console.log(
|
||||
`[Maestro] Prepare quorum armed: ${this.expected.size} participant(s) ` +
|
||||
`(agents + primordial daemons), timeout ${this.timeoutMs}ms`
|
||||
)
|
||||
}
|
||||
|
||||
return(new Promise(resolve => {
|
||||
this.resolve = resolve
|
||||
this.timer = setTimeout(() => {
|
||||
const missing = [...this.expected].filter(id => !this.ready.has(id))
|
||||
this.#finish({
|
||||
ok: false,
|
||||
err: `Timeout waiting for readyToStart (${this.timeoutMs}ms); ` +
|
||||
`missing: ${missing.join(', ') || 'unknown'}`,
|
||||
})
|
||||
}, this.timeoutMs)
|
||||
}))
|
||||
}
|
||||
|
||||
handleMessage(msg, chan) {
|
||||
if(!this.active) return(false)
|
||||
if(typeof(this.matchesChan) !== 'function') return(false)
|
||||
if(!this.matchesChan(chan, this.ackChannel)) return(false)
|
||||
if(msg?.eventType !== 'readyToStart') return(false)
|
||||
|
||||
const payload = msg.payload ?? {}
|
||||
if(payload.simulationId !== this.simulationId) return(false)
|
||||
|
||||
const sender = msg.sender
|
||||
if(!sender || !this.expected.has(sender)) {
|
||||
if(this.debug) {
|
||||
console.warn(`[Maestro] Ignoring readyToStart from unexpected participant: ${sender}`)
|
||||
}
|
||||
return(true)
|
||||
}
|
||||
|
||||
if(!payload.success) {
|
||||
this.#finish({
|
||||
ok: false,
|
||||
err: payload.err ?? `Participant ${sender} failed prepare`,
|
||||
})
|
||||
return(true)
|
||||
}
|
||||
|
||||
this.ready.set(sender, payload)
|
||||
if(this.debug) {
|
||||
console.log(
|
||||
`[Maestro] readyToStart from ${sender} ` +
|
||||
`(${this.ready.size}/${this.expected.size})`
|
||||
)
|
||||
}
|
||||
|
||||
for(const participantId of this.expected) {
|
||||
if(!this.ready.has(participantId)) return(true)
|
||||
}
|
||||
|
||||
this.#finish({ ok: true })
|
||||
return(true)
|
||||
}
|
||||
|
||||
cancel() {
|
||||
if(this.timer) {
|
||||
clearTimeout(this.timer)
|
||||
this.timer = null
|
||||
}
|
||||
this.active = false
|
||||
this.resolve = null
|
||||
this.expected.clear()
|
||||
this.ready.clear()
|
||||
this.simulationId = null
|
||||
}
|
||||
|
||||
#finish(result) {
|
||||
const resolve = this.resolve
|
||||
this.cancel()
|
||||
if(typeof(resolve) === 'function') resolve(result)
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
const SKIP_PRIMORDIAL_SECTIONS = new Set([
|
||||
'maestro',
|
||||
'mysql',
|
||||
'accessRights',
|
||||
'systemMesh',
|
||||
'arenaMesh',
|
||||
])
|
||||
|
||||
export function getPrimordialDaemonIds(config = {}) {
|
||||
const ids = []
|
||||
for(const [section, block] of Object.entries(config)) {
|
||||
if(SKIP_PRIMORDIAL_SECTIONS.has(section)) continue
|
||||
if(!block || typeof(block) !== 'object' || Array.isArray(block)) continue
|
||||
if(!block.primordialDaemon) continue
|
||||
ids.push(typeof(block.senderId) === 'string' ? block.senderId : section)
|
||||
}
|
||||
return(ids)
|
||||
}
|
||||
|
||||
export function buildPrepareQuorum(agentIds, config) {
|
||||
const primordialIds = getPrimordialDaemonIds(config)
|
||||
return([...agentIds, ...primordialIds])
|
||||
}
|
||||
@@ -0,0 +1,140 @@
|
||||
import { MySQLClient } from '@p42/p42modules'
|
||||
|
||||
const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i
|
||||
|
||||
export function isValidUuid(val) {
|
||||
return(typeof(val) === 'string' && UUID_RE.test(val))
|
||||
}
|
||||
|
||||
export class SimRepository {
|
||||
|
||||
constructor(dbPool, databases, debug = false) {
|
||||
this.db = dbPool
|
||||
this.guiDb = databases.guiDatabase
|
||||
this.simDb = databases.simDatabase
|
||||
this.debug = debug
|
||||
}
|
||||
|
||||
#qualify(db, table) {
|
||||
return(`\`${db}\`.${table}`)
|
||||
}
|
||||
|
||||
#parseJsonObject(raw) {
|
||||
let v = raw
|
||||
if(v == null) return(null)
|
||||
if(typeof(v) === 'string') {
|
||||
try { v = JSON.parse(v) }
|
||||
catch { return(null) }
|
||||
}
|
||||
if(typeof(v) !== 'object' || Array.isArray(v)) return(null)
|
||||
return(v)
|
||||
}
|
||||
|
||||
#parseGpsValues(raw) {
|
||||
const v = this.#parseJsonObject(raw)
|
||||
if(!v) return(null)
|
||||
const position = v.position
|
||||
const speed = v.speed ?? v.vector
|
||||
if(!position || !speed) return(null)
|
||||
const axes = ['x', 'y', 'z']
|
||||
for(const axis of axes) {
|
||||
if(typeof(position[axis]) !== 'number' || !Number.isFinite(position[axis])) return(null)
|
||||
if(typeof(speed[axis]) !== 'number' || !Number.isFinite(speed[axis])) return(null)
|
||||
}
|
||||
return({
|
||||
position: { x: position.x, y: position.y, z: position.z },
|
||||
vector: { x: speed.x, y: speed.y, z: speed.z },
|
||||
})
|
||||
}
|
||||
|
||||
#parseStoreValues(raw) {
|
||||
const v = this.#parseJsonObject(raw)
|
||||
if(v == null) return(null)
|
||||
return({ ...v })
|
||||
}
|
||||
|
||||
async validateSimulationAccess(userUuid, simulationUuid) {
|
||||
if(!isValidUuid(userUuid)) return({ ok: false, err: 'Invalid user UUID' })
|
||||
if(!isValidUuid(simulationUuid)) return({ ok: false, err: 'Invalid simulation UUID' })
|
||||
|
||||
const rows = await MySQLClient.poolExecute(this.db, `
|
||||
SELECT s.sim_id,
|
||||
BIN_TO_UUID(s.sim_uuid) AS sim_uuid,
|
||||
BIN_TO_UUID(s.sim_root_kf_uuid) AS sim_root_kf_uuid
|
||||
FROM ${this.#qualify(this.simDb, 'simulations')} s
|
||||
INNER JOIN ${this.#qualify(this.guiDb, 'simowners')} o ON o.own_sim_uuid = s.sim_uuid
|
||||
INNER JOIN ${this.#qualify(this.guiDb, 'users')} u ON o.own_usr_id = u.usr_id
|
||||
WHERE u.usr_uuid = ?
|
||||
AND s.sim_uuid = UUID_TO_BIN(?)
|
||||
`, [userUuid, simulationUuid])
|
||||
|
||||
if(!rows.length) return({ ok: false, err: 'Simulation not found or access denied' })
|
||||
|
||||
const sim = rows[0]
|
||||
if(!sim.sim_root_kf_uuid) return({ ok: false, err: 'Simulation has no root keyframe' })
|
||||
|
||||
const kfRows = await MySQLClient.poolExecute(this.db, `
|
||||
SELECT ekf_uuid
|
||||
FROM ${this.#qualify(this.simDb, 'edited_keyframes')}
|
||||
WHERE ekf_uuid = UUID_TO_BIN(?)
|
||||
`, [sim.sim_root_kf_uuid])
|
||||
if(!kfRows.length) return({ ok: false, err: 'Root keyframe not found' })
|
||||
|
||||
return({ ok: true, sim })
|
||||
}
|
||||
|
||||
async validateSimulationOwner(userUuid, simulationUuid) {
|
||||
if(!isValidUuid(userUuid)) return({ ok: false, err: 'Invalid user UUID' })
|
||||
if(!isValidUuid(simulationUuid)) return({ ok: false, err: 'Invalid simulation UUID' })
|
||||
|
||||
const rows = await MySQLClient.poolExecute(this.db, `
|
||||
SELECT s.sim_id,
|
||||
BIN_TO_UUID(s.sim_uuid) AS sim_uuid,
|
||||
BIN_TO_UUID(s.sim_root_kf_uuid) AS sim_root_kf_uuid
|
||||
FROM ${this.#qualify(this.simDb, 'simulations')} s
|
||||
INNER JOIN ${this.#qualify(this.guiDb, 'simowners')} o ON o.own_sim_uuid = s.sim_uuid
|
||||
INNER JOIN ${this.#qualify(this.guiDb, 'users')} u ON o.own_usr_id = u.usr_id
|
||||
WHERE u.usr_uuid = ?
|
||||
AND s.sim_uuid = UUID_TO_BIN(?)
|
||||
`, [userUuid, simulationUuid])
|
||||
|
||||
if(!rows.length) return({ ok: false, err: 'Simulation not found or access denied' })
|
||||
return({ ok: true, sim: rows[0] })
|
||||
}
|
||||
|
||||
async loadKeyframeAgents(keyframeId) {
|
||||
const rows = await MySQLClient.poolExecute(this.db, `
|
||||
SELECT BIN_TO_UUID(ekfs_agent_id) AS agent_id,
|
||||
ekfs_gps_values,
|
||||
ekfs_store_values
|
||||
FROM ${this.#qualify(this.simDb, 'edited_kf_store')}
|
||||
WHERE ekfs_ekf_uuid = UUID_TO_BIN(?)
|
||||
`, [keyframeId])
|
||||
|
||||
const agents = []
|
||||
const errors = []
|
||||
|
||||
for(const row of rows) {
|
||||
const parsed = this.#parseGpsValues(row.ekfs_gps_values)
|
||||
if(!parsed) {
|
||||
errors.push(`Invalid GPS values for agent ${row.agent_id}`)
|
||||
continue
|
||||
}
|
||||
const store = this.#parseStoreValues(row.ekfs_store_values)
|
||||
if(store == null) {
|
||||
errors.push(`Invalid store values for agent ${row.agent_id}`)
|
||||
continue
|
||||
}
|
||||
agents.push({
|
||||
id: row.agent_id,
|
||||
position: parsed.position,
|
||||
vector: parsed.vector,
|
||||
store,
|
||||
})
|
||||
}
|
||||
|
||||
if(errors.length) return({ ok: false, err: errors.join('; '), agents: [] })
|
||||
return({ ok: true, agents })
|
||||
}
|
||||
|
||||
}
|
||||
Executable
+33
@@ -0,0 +1,33 @@
|
||||
#!/bin/sh
|
||||
|
||||
set -a
|
||||
. /etc/p42/secrets.env
|
||||
set +a
|
||||
|
||||
daemon=p42Maestro
|
||||
logfile=maestro.log
|
||||
|
||||
pid=$(pgrep -f "$daemon")
|
||||
|
||||
if [ -z "$pid" ]
|
||||
then
|
||||
node "${daemon}.js" --debug > "$logfile" 2>&1 &
|
||||
pid=$!
|
||||
|
||||
sleep 1
|
||||
|
||||
if kill -0 "$pid" 2>/dev/null
|
||||
then
|
||||
echo ""
|
||||
echo "$daemon is now running with PID=$pid"
|
||||
echo ""
|
||||
else
|
||||
echo ""
|
||||
echo "Failed to start $daemon. Check maestro.log"
|
||||
echo ""
|
||||
fi
|
||||
else
|
||||
echo ""
|
||||
echo "$daemon is already running with PID=$pid"
|
||||
echo ""
|
||||
fi
|
||||
Executable
+8
@@ -0,0 +1,8 @@
|
||||
#!/bin/sh
|
||||
|
||||
pid=`ps -ef | grep p42Maestro.js |grep -v grep | awk '{print $2}'`
|
||||
if [ -n "$pid" ]
|
||||
then
|
||||
echo "killing pid: $pid"
|
||||
kill -9 $pid
|
||||
fi
|
||||
Reference in New Issue
Block a user