diff --git a/servers/mu/src/domain/api/pushResultToHb.js b/servers/mu/src/domain/api/pushResultToHb.js new file mode 100644 index 000000000..64e14b1e9 --- /dev/null +++ b/servers/mu/src/domain/api/pushResultToHb.js @@ -0,0 +1,72 @@ +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, + ENABLE_PUSH, + 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) => { + 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 })) + } + 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/clients/uploader.js b/servers/mu/src/domain/clients/uploader.js index e169731e3..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,26 +23,22 @@ 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}` })) - .chain( - fromPromise((body) => - 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`, { + 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', @@ -49,25 +46,22 @@ function uploadDataItemWith ({ UPLOADER_URL, fetch, histogram, logger, HB_GRAPHQ }, body }) - .then(() => res) - .catch((err) => { - logger.tap({ log: 'Error while communicating with HB uploader:' })(err) - return 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() } } diff --git a/servers/mu/src/domain/index.js b/servers/mu/src/domain/index.js index 3e9629940..317d0a58d 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,17 @@ 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, + ENABLE_PUSH: ctx.ENABLE_PUSH, + fetch + }) + const pushMsgItemLogger = logger.child('pushMsg') const pushMsg = pushMsgWith({ selectNode: cuClient.selectNodeWith({ CU_URL, logger: sendDataItemLogger }), @@ -447,6 +459,7 @@ export const createApis = async (ctx) => { sendAssign, fetchCron, pushMsg, + pushResultToHb, traceMsgs, initCronProcs: cronClient.initCronProcsWith({ startMonitoredProcess: startProcessMonitor, 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) } 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 +)