diff --git a/servers/cu/src/domain/lib/evaluate.js b/servers/cu/src/domain/lib/evaluate.js index ce93bfb11..236b3e9a0 100644 --- a/servers/cu/src/domain/lib/evaluate.js +++ b/servers/cu/src/domain/lib/evaluate.js @@ -344,12 +344,15 @@ export function evaluateWith (env) { const now = new Date() // If there is no startTime, then we use the current time which will make evalTime = 0 const startTime = pathOr(now, ['stats', 'startTime'], ctx) - + const assignmentIdToSave = + (message.Skip === 'true' && message['Original-Assignment-Id']) + ? message['Original-Assignment-Id'] + : mostRecentAssignmentId await saveLatestProcessMemory({ processId: ctx.id, moduleId: ctx.moduleId, messageId: message.Id, - assignmentId: mostRecentAssignmentId, + assignmentId: assignmentIdToSave, hashChain: mostRecentHashChain, timestamp: message.Timestamp, nonce: message.Nonce, diff --git a/servers/cu/src/domain/model.js b/servers/cu/src/domain/model.js index bb8b2dd62..c7e57cf51 100644 --- a/servers/cu/src/domain/model.js +++ b/servers/cu/src/domain/model.js @@ -371,6 +371,7 @@ export const messageSchema = z.object({ */ Cron: z.boolean(), Skip: z.string().default('false'), + 'Original-Assignment-Id': z.string().nullish(), 'Read-Only': z.boolean().default(false) }), AoGlobal: z.object({ diff --git a/servers/cu/src/effects/hb/index.js b/servers/cu/src/effects/hb/index.js index 49cdd4c59..af5c55e03 100644 --- a/servers/cu/src/effects/hb/index.js +++ b/servers/cu/src/effects/hb/index.js @@ -183,6 +183,7 @@ export const mapNode = (type) => pipe( Epoch: pipe(path(['Epoch']), parseInt), Nonce: pipe(path(['Slot'])(tags) ? path(['Slot']) : path(['Nonce']), parseInt), Skip: pathOr('false', ['Skip']), + 'Original-Assignment-Id': path(['Original-Assignment-Id']), Timestamp: pipe(path(['Timestamp']), parseInt), 'Block-Height': pipe( /** @@ -454,6 +455,7 @@ export const loadMessagesWith = ({ hashChain, fetch, logger: _logger, pageSize } body.edges.length === (+to - +from + 1) && +body.edges[0]?.node?.assignment?.Tags?.find(t => t.name === 'Nonce' || t.name === 'Slot')?.value === +from if (bodyIsValid) return body + if (!dryRun) throw new Error('Body is not valid: would attempt to fetch from scheduler in loadMessages') return fetchPageDataloader.load({ suUrl, processId, from, to, pageSize }) }, { maxRetries: 1, delay: 500, log: logger, name: `loadMessages(${JSON.stringify({ suUrl, processId, params: params.toString() })})` } @@ -570,7 +572,11 @@ export const loadMessagesWith = ({ hashChain, fetch, logger: _logger, pageSize } } expectedNonce = expectedNonce + 1 - prevAssignmentId = scheduled.assignmentId + if (scheduled.message.Skip === 'true' && scheduled.message['Original-Assignment-Id']) { + prevAssignmentId = scheduled.message['Original-Assignment-Id'] + } else { + prevAssignmentId = scheduled.assignmentId + } prevHashChain = scheduled.message['Hash-Chain'] yield scheduled @@ -612,7 +618,7 @@ export const loadMessageMetaWith = ({ fetch, logger }) => { body.edges.length === 1 && +body.edges[0]?.node?.assignment?.Tags?.find(t => t.name === 'Nonce' || t.name === 'Slot')?.value === +messageUid if (bodyIsValid) return body - return fetch(`${suUrl}/~scheduler@1.0/schedule?${params.toString()}`).then(okRes) + throw new Error('Body is not valid: would attempt to fetch from scheduler in loadMessageMeta') }, { maxRetries: 2, delay: 500, log: logger, name: `loadMessageMeta(${JSON.stringify({ suUrl, processId, messageUid })})` } )