From 7d5290970127826be76d9e02a5a678ef7244912c Mon Sep 17 00:00:00 2001 From: Jack Frain Date: Tue, 3 Mar 2026 19:15:09 -0500 Subject: [PATCH 1/7] chore(mu): add hb upload log --- servers/mu/src/domain/clients/uploader.js | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/servers/mu/src/domain/clients/uploader.js b/servers/mu/src/domain/clients/uploader.js index e169731e3..1d386191b 100644 --- a/servers/mu/src/domain/clients/uploader.js +++ b/servers/mu/src/domain/clients/uploader.js @@ -49,7 +49,10 @@ function uploadDataItemWith ({ UPLOADER_URL, fetch, histogram, logger, HB_GRAPHQ }, body }) - .then(() => res) + .then((HBRes) => { + logger.tap({ log: 'Successfully forwarded DataItem to HB uploader' })(HBRes) + return res + }) .catch((err) => { logger.tap({ log: 'Error while communicating with HB uploader:' })(err) return res From 0aab0f5f37239a7aba2b536d96da40c87025786c Mon Sep 17 00:00:00 2001 From: Jack Frain Date: Tue, 3 Mar 2026 19:16:52 -0500 Subject: [PATCH 2/7] chore(mu): add hb upload log --- servers/mu/src/domain/clients/uploader.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/servers/mu/src/domain/clients/uploader.js b/servers/mu/src/domain/clients/uploader.js index 1d386191b..4ca4311c2 100644 --- a/servers/mu/src/domain/clients/uploader.js +++ b/servers/mu/src/domain/clients/uploader.js @@ -26,7 +26,7 @@ function uploadDataItemWith ({ UPLOADER_URL, fetch, histogram, logger, HB_GRAPHQ */ return async (data) => { return of(data) - .map(logger.tap({ log: `Forwarding message to uploader ${UPLOADER_URL}` })) + .map(logger.tap({ log: `Forwarding message to uploader ${UPLOADER_URL}, HB uploader ${HB_GRAPHQL_URL}` })) .chain( fromPromise((body) => dataItemFetch(`${UPLOADER_URL}/tx/arweave`, { From 2d83311b00daf50bf02eaf7d300ec1f7388198a1 Mon Sep 17 00:00:00 2001 From: Jack Frain Date: Tue, 3 Mar 2026 20:46:52 -0500 Subject: [PATCH 3/7] chore(mu): add hb upload log --- servers/mu/src/domain/clients/uploader.js | 11 ++--------- 1 file changed, 2 insertions(+), 9 deletions(-) diff --git a/servers/mu/src/domain/clients/uploader.js b/servers/mu/src/domain/clients/uploader.js index 4ca4311c2..2a3d36650 100644 --- a/servers/mu/src/domain/clients/uploader.js +++ b/servers/mu/src/domain/clients/uploader.js @@ -43,20 +43,13 @@ function uploadDataItemWith ({ UPLOADER_URL, fetch, histogram, logger, HB_GRAPHQ fromPromise(({ body, res }) => dataItemFetch(`${HB_GRAPHQL_URL}/~arweave@2.9-pre/tx?codec-device=ans104@1.0`, { method: 'POST', - headers: { - 'Content-Type': 'application/octet-stream', - Accept: 'application/json' - }, body }) .then((HBRes) => { - logger.tap({ log: 'Successfully forwarded DataItem to HB uploader' })(HBRes) - return res - }) - .catch((err) => { - logger.tap({ log: 'Error while communicating with HB uploader:' })(err) + logger.tap({ log: 'Successfully forwarded DataItem to HB uploader' })() return res }) + .catch((_) => res) ) ) .bimap(logger.tap({ log: 'Error while communicating with uploader:' }), identity) From bc57625e9f985bcc9df4f48e60c6c0823aeaba2e Mon Sep 17 00:00:00 2001 From: Jack Frain Date: Thu, 5 Mar 2026 17:20:49 -0500 Subject: [PATCH 4/7] fix(mu): improve hb upload flow --- servers/mu/src/domain/clients/uploader.js | 42 +++++++++++------------ 1 file changed, 20 insertions(+), 22 deletions(-) diff --git a/servers/mu/src/domain/clients/uploader.js b/servers/mu/src/domain/clients/uploader.js index 2a3d36650..037573139 100644 --- a/servers/mu/src/domain/clients/uploader.js +++ b/servers/mu/src/domain/clients/uploader.js @@ -11,9 +11,10 @@ function uploadDataItemWith ({ UPLOADER_URL, fetch, histogram, logger, HB_GRAPHQ }), logger }) + /** * uploadDataItem - * Upload a Data Item directly to Arweave + * Upload a Data Item directly to Arweave, and fire-and-forget to HB. * * @param data - the Data Item to upload * @@ -22,48 +23,45 @@ function uploadDataItemWith ({ UPLOADER_URL, fetch, histogram, logger, HB_GRAPHQ * timestamp * signature * owner - * */ return async (data) => { return of(data) - .map(logger.tap({ log: `Forwarding message to uploader ${UPLOADER_URL}, HB uploader ${HB_GRAPHQL_URL}` })) .chain( - fromPromise((body) => - dataItemFetch(`${UPLOADER_URL}/tx/arweave`, { + fromPromise(async (body) => { + // Fire HB upload in parallel — never blocks or fails the main flow + const hbUrl = `${HB_GRAPHQL_URL}/id?codec-device=ans104@1.0` + logger.tap({ log: `[uploader] Forwarding to HB: ${hbUrl}` })() + dataItemFetch(hbUrl, { method: 'POST', body }) + .then((res) => res.text()) + .then((text) => logger.tap({ log: `[uploader] HB response: ${text}` })()) + .catch((err) => logger.tap({ log: `[uploader] HB upload error (non-fatal): ${err.message}` })()) + + // Arweave upload — this is the one we await and return + logger.tap({ log: `[uploader] Forwarding to Arweave: ${UPLOADER_URL}/tx/arweave` })() + return dataItemFetch(`${UPLOADER_URL}/tx/arweave`, { method: 'POST', headers: { 'Content-Type': 'application/octet-stream', Accept: 'application/json' }, body - }).then((res) => { return { body, res } }) - ) - ) - .chain( - fromPromise(({ body, res }) => - dataItemFetch(`${HB_GRAPHQL_URL}/~arweave@2.9-pre/tx?codec-device=ans104@1.0`, { - method: 'POST', - body }) - .then((HBRes) => { - logger.tap({ log: 'Successfully forwarded DataItem to HB uploader' })() - return res - }) - .catch((_) => res) - ) + }) ) - .bimap(logger.tap({ log: 'Error while communicating with uploader:' }), identity) + .bimap(logger.tap({ log: '[uploader] Error communicating with Arweave uploader:' }), identity) .bichain( (err) => Rejected(JSON.stringify(err)), fromPromise(async (res) => { if (!res?.ok) { const text = await res.text() + logger.tap({ log: `[uploader] Arweave upload failed: ${res.status} ${text}` })() throw new Error(`${res.status}: ${text}`) } - return res.json() + const json = await res.json() + logger.tap({ log: `[uploader] Arweave upload succeeded: ${json.id}` })() + return json }) ) - .map(logger.tap({ log: 'Successfully forwarded DataItem to uploader' })) .toPromise() } } From 775db5bde27431763a2c8983958f0a206c840ab3 Mon Sep 17 00:00:00 2001 From: Jack Frain Date: Thu, 5 Mar 2026 17:21:14 -0500 Subject: [PATCH 5/7] fix(mu): disable cu rdirect cache --- servers/mu/src/domain/lib/cu-fetch-with-cache.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/servers/mu/src/domain/lib/cu-fetch-with-cache.js b/servers/mu/src/domain/lib/cu-fetch-with-cache.js index d58cc1073..53b563663 100644 --- a/servers/mu/src/domain/lib/cu-fetch-with-cache.js +++ b/servers/mu/src/domain/lib/cu-fetch-with-cache.js @@ -34,7 +34,7 @@ export function cuFetchWithCache ({ fetch, cache, logger }) { logger({ log: ['found redirect url in cache for process: %s redirect: %s', processId, foundRedirectUrl], logId }) // only sets the host, the protocol will be reused from the passed in url // this is a safe assumption because all CUs are implementing the same APIs over the same protocols - requestUrl.host = foundRedirectUrl + // requestUrl.host = foundRedirectUrl } return runFetch(requestUrl.toString(), opts, logId) } From ecbec940012b27cb030ab9b54e9ebf075f537c50 Mon Sep 17 00:00:00 2001 From: Jack Frain Date: Thu, 5 Mar 2026 17:54:30 -0500 Subject: [PATCH 6/7] feat(mu): add push msg to hb route --- servers/mu/src/domain/api/pushResultToHb.js | 68 +++++++++++++++++++++ servers/mu/src/domain/index.js | 12 ++++ servers/mu/src/routes/index.js | 4 +- servers/mu/src/routes/pushResultToHb.js | 58 ++++++++++++++++++ 4 files changed, 141 insertions(+), 1 deletion(-) create mode 100644 servers/mu/src/domain/api/pushResultToHb.js create mode 100644 servers/mu/src/routes/pushResultToHb.js diff --git a/servers/mu/src/domain/api/pushResultToHb.js b/servers/mu/src/domain/api/pushResultToHb.js new file mode 100644 index 000000000..d0c8eb670 --- /dev/null +++ b/servers/mu/src/domain/api/pushResultToHb.js @@ -0,0 +1,68 @@ +import { Rejected, Resolved, fromPromise, of } from 'hyper-async' + +import { getCuAddressWith } from '../lib/get-cu-address.js' +import { pullResultWith } from '../lib/pull-result.js' + +export function pushResultToHbWith ({ + selectNode, + fetchResult, + buildAndSign, + logger, + HB_GRAPHQL_URL, + fetch +}) { + const getCuAddress = getCuAddressWith({ selectNode, logger }) + const pullResult = pullResultWith({ fetchResult, logger }) + const buildAndSignAsync = fromPromise(buildAndSign) + + const uploadToHb = async ({ signedDataItem, processId, messageId, logId }) => { + const url = `${HB_GRAPHQL_URL}/id?codec-device=ans104@1.0` + logger({ log: `[pushResultToHb] Uploading signed data item to HB: ${url} processId=${processId} messageId=${messageId} logId=${logId}` }) + const res = await fetch(url, { + method: 'POST', + headers: { 'Content-Type': 'application/octet-stream' }, + body: signedDataItem + }) + const text = await res.text() + if (!res.ok) { + throw new Error(`[pushResultToHb] HB upload failed: ${res.status} ${text}`) + } + logger({ log: `[pushResultToHb] HB upload succeeded: ${text}` }) + return text + } + + const uploadToHbAsync = fromPromise(uploadToHb) + + return (ctx) => { + return of(ctx) + .chain(getCuAddress) + .chain(pullResult) + .chain((res) => { + const { msgs, number } = res + if (msgs.length <= number) { + return Rejected(new Error('Message number does not exist in the result.', { cause: ctx })) + } + return Resolved(res) + }) + .chain((res) => { + const { msgs, number } = res + const targetMsg = msgs[number].msg + logger({ log: `[pushResultToHb] Building and signing result message ${number} for ${ctx.tx.id} -> target=${targetMsg.Target}` }) + console.dir({ targetMsg }, { depth: null }) + return buildAndSignAsync({ + processId: targetMsg.Target, + tags: targetMsg.Tags, + anchor: targetMsg.Anchor, + data: targetMsg.Data + }).chain((tx) => { + logger({ log: `[pushResultToHb] Signed data item id=${tx.id}, uploading to HB` }) + return uploadToHbAsync({ + signedDataItem: tx.data, + processId: ctx.tx.processId, + messageId: ctx.tx.id, + logId: ctx.logId + }).map((hbRes) => ({ ...res, hbRes, txId: tx.id })) + }) + }) + } +} diff --git a/servers/mu/src/domain/index.js b/servers/mu/src/domain/index.js index 3e9629940..64abca926 100644 --- a/servers/mu/src/domain/index.js +++ b/servers/mu/src/domain/index.js @@ -29,6 +29,7 @@ import { sendDataItemWith, startMessageRecoveryCronWith } from './api/sendDataIt import { sendAssignWith } from './api/sendAssign.js' import { processAssignWith } from './api/processAssign.js' import { pushMsgWith } from './api/pushMsg.js' +import { pushResultToHbWith } from './api/pushResultToHb.js' import { createLogger } from './logger.js' import { cuFetchWithCache } from './lib/cu-fetch-with-cache.js' @@ -410,6 +411,16 @@ export const createApis = async (ctx) => { const traceMsgs = fromPromise(readTracesWith({ db: traceDb, TRACE_DB_URL: ctx.TRACE_DB_URL, DISABLE_TRACE: ctx.DISABLE_TRACE })) + const pushResultToHbLogger = logger.child('pushResultToHb') + const pushResultToHb = pushResultToHbWith({ + selectNode: cuClient.selectNodeWith({ CU_URL, logger: pushResultToHbLogger }), + fetchResult: cuClient.resultWith({ fetch: fetchWithCache, histogram, CU_URL, logger: pushResultToHbLogger }), + buildAndSign: signerClient.buildAndSignWith({ MU_WALLET, logger: pushResultToHbLogger }), + logger: pushResultToHbLogger, + HB_GRAPHQL_URL, + fetch + }) + const pushMsgItemLogger = logger.child('pushMsg') const pushMsg = pushMsgWith({ selectNode: cuClient.selectNodeWith({ CU_URL, logger: sendDataItemLogger }), @@ -447,6 +458,7 @@ export const createApis = async (ctx) => { sendAssign, fetchCron, pushMsg, + pushResultToHb, traceMsgs, initCronProcs: cronClient.initCronProcsWith({ startMonitoredProcess: startProcessMonitor, diff --git a/servers/mu/src/routes/index.js b/servers/mu/src/routes/index.js index 4364a3828..77f015a10 100644 --- a/servers/mu/src/routes/index.js +++ b/servers/mu/src/routes/index.js @@ -4,10 +4,12 @@ import { withRootRoutes } from './root.js' import { withMonitorRoutes } from './monitor.js' import { withMetricRoutes } from './metrics.js' import { withPushRoutes } from './push.js' +import { withPushResultToHbRoutes } from './pushResultToHb.js' export const withRoutes = pipe( withMonitorRoutes, withRootRoutes, withMetricRoutes, - withPushRoutes + withPushRoutes, + withPushResultToHbRoutes ) diff --git a/servers/mu/src/routes/pushResultToHb.js b/servers/mu/src/routes/pushResultToHb.js new file mode 100644 index 000000000..5245c85d3 --- /dev/null +++ b/servers/mu/src/routes/pushResultToHb.js @@ -0,0 +1,58 @@ +import { always, compose, pipe } from 'ramda' +import { of } from 'hyper-async' +import { randomBytes } from 'node:crypto' + +import { withMetrics, withMiddleware } from './middleware/index.js' + +const withPushResultToHbRoute = (app) => { + app.post( + '/push-result/:id/:number', + compose( + withMiddleware, + withMetrics(), + always(async (req, res) => { + const { + logger: _logger, + domain: { apis: { pushResultToHb } }, + params: { id, number }, + query: { + 'process-id': processId + } + } = req + + const logger = _logger.child('POST_push_result_to_hb') + const logId = randomBytes(8).toString('hex') + + if (isNaN(Number(number))) { + return res.status(400).send({ error: "'number' parameter must be a valid number" }) + } + + await of({ + tx: { id, processId }, + number: Number(number), + logId, + messageId: id, + initialTxId: id + }) + .chain(pushResultToHb) + .bimap( + (e) => { + logger({ log: `[push-result] Failed: ${e}`, end: true }, e.cause) + res.status(500).send({ error: String(e) }) + }, + ({ hbRes }) => { + logger({ log: `[push-result] Success for ${id}/${number}`, end: true }) + res.status(200).send({ message: 'Result uploaded to HB', id, number: Number(number), hbRes }) + } + ) + .toPromise() + }) + )() + ) + + return app +} + +export const withPushResultToHbRoutes = pipe( + withPushResultToHbRoute +) From dc3540b5a0f5203c0480d50082509d1ff93fe4b7 Mon Sep 17 00:00:00 2001 From: VinceJuliano Date: Thu, 5 Mar 2026 18:58:48 -0500 Subject: [PATCH 7/7] chore(mu): only allow push to cache if push enabled --- servers/mu/src/domain/api/pushResultToHb.js | 4 ++++ servers/mu/src/domain/index.js | 1 + 2 files changed, 5 insertions(+) diff --git a/servers/mu/src/domain/api/pushResultToHb.js b/servers/mu/src/domain/api/pushResultToHb.js index d0c8eb670..64e14b1e9 100644 --- a/servers/mu/src/domain/api/pushResultToHb.js +++ b/servers/mu/src/domain/api/pushResultToHb.js @@ -9,6 +9,7 @@ export function pushResultToHbWith ({ buildAndSign, logger, HB_GRAPHQL_URL, + ENABLE_PUSH, fetch }) { const getCuAddress = getCuAddressWith({ selectNode, logger }) @@ -38,6 +39,9 @@ export function pushResultToHbWith ({ .chain(getCuAddress) .chain(pullResult) .chain((res) => { + if(!ENABLE_PUSH) { + return Rejected(new Error('Repush not enabled on this MU.', { cause: ctx })) + } const { msgs, number } = res if (msgs.length <= number) { return Rejected(new Error('Message number does not exist in the result.', { cause: ctx })) diff --git a/servers/mu/src/domain/index.js b/servers/mu/src/domain/index.js index 64abca926..317d0a58d 100644 --- a/servers/mu/src/domain/index.js +++ b/servers/mu/src/domain/index.js @@ -418,6 +418,7 @@ export const createApis = async (ctx) => { buildAndSign: signerClient.buildAndSignWith({ MU_WALLET, logger: pushResultToHbLogger }), logger: pushResultToHbLogger, HB_GRAPHQL_URL, + ENABLE_PUSH: ctx.ENABLE_PUSH, fetch })