feat(cu): use original assignment id to deduped messages

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,

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({

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 })})` }
     )