Skip to content

Commit 441e137

Browse files
guidiazguidiaz
committed
feat(notarizer): evaluate heartbeat condition before attemptying dry run
1 parent a0e7edb commit 441e137

1 file changed

Lines changed: 130 additions & 121 deletions

File tree

‎src/bin/bots/notarizer.cjs‎

Lines changed: 130 additions & 121 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ async function main() {
5151
.option(
5252
"--max-threads <number>",
5353
"Max. number of simultaneous dry runs",
54-
process.env.WITNET_PFS_DRY_RUN_MAX_THREADS || 1,
54+
process.env.WITNET_PFS_DRY_RUN_MAX_THREADS || 2,
5555
)
5656
.option(
5757
"--min-balance <wits>",
@@ -218,139 +218,148 @@ async function main() {
218218
}
219219
}
220220

221-
// prepare and spawn new dry-run subprocess:
222-
const dryRunStart = Date.now();
223-
metrics.dryruns += 1;
224-
console.debug(`[${tag}] Dry-running ${lastDryRunClock ? `after ${commas(dryRunStart - lastDryRunClock)} msecs` : `for the first time`} ...`);
225-
priceFeeds[caption].lastDryRunClock = dryRunStart;
226-
threadBucket.push(
227-
request.execDryRun({ timeout: DRY_RUN_TIMEOUT_SECS * 1000 })
228-
.then(output => {
229-
if (!output || output === "") throw new Error(`no dry-run report`);
230-
else return JSON.parse(output);
231-
})
232-
.then(json => {
233-
// parse dry run result
234-
console.debug(`[${tag}] Dry-run solved in ${commas(Date.now() - dryRunStart)} msecs => ${JSON.stringify(json)}`);
235-
if (!Object.keys(json).includes("RadonInteger")) {
236-
throw `Error: unexpected dry run result: ${JSON.stringify(json).slice(0, 2048)}`;
237-
}
238-
const currentValue = parseInt(json.RadonInteger, 10);
239-
240-
// determine whether a new notarization is required
241-
const heartbeatSecs = Math.floor(Date.now() / 1000) - lastUpdates[caption].timestamp;
242-
if (heartbeatSecs < conditions.heartbeatSecs / 2 + 1) {
243-
const deviation =
244-
lastUpdates[caption].value > 0
245-
? (100 * (currentValue - lastUpdates[caption].value)) /
246-
lastUpdates[caption].value
247-
: 0;
248-
if (Math.abs(deviation) < conditions.deviationPercentage) {
249-
console.info(
250-
`[${tag}] ${deviation >= 0 ? "+" : ""}${deviation.toFixed(2)} % deviation after ${heartbeatSecs} secs.`
251-
)
252-
return;
253-
} else {
254-
console.info(
255-
`[${tag}] Updating due to price deviation of ${deviation.toFixed(2)} % ...`,
256-
);
257-
}
221+
threadBucket.push(new Promise(async (resolve, reject) => {
222+
223+
// determine whether a new notarization is required
224+
const heartbeatSecs = Math.floor(Date.now() / 1000) - lastUpdates[caption].timestamp;
225+
if (heartbeatSecs < conditions.heartbeatSecs / 2 + 1) {
226+
// prepare dry-run subprocess
227+
const dryRunStart = Date.now();
228+
metrics.dryruns += 1;
229+
console.debug(`[${tag}] Dry-running ${lastDryRunClock ? `after ${commas(dryRunStart - lastDryRunClock)} msecs` : `for the first time`} ...`);
230+
priceFeeds[caption].lastDryRunClock = dryRunStart;
231+
232+
// determine current market value
233+
const currentValue = await request.execDryRun({ timeout: DRY_RUN_TIMEOUT_SECS * 1000 })
234+
.then(output => {
235+
if (!output || output === "") throw new Error(`no dry-run report`);
236+
else return JSON.parse(output);
237+
})
238+
.then(json => {
239+
// parse dry run result
240+
console.debug(`[${tag}] Dry-run solved in ${commas(Date.now() - dryRunStart)} msecs => ${JSON.stringify(json)}`);
241+
if (!Object.keys(json).includes("RadonInteger")) {
242+
throw new Error(`unexpected dry run result: ${JSON.stringify(json).slice(0, 2048)}`);
243+
} else {
244+
return parseInt(json.RadonInteger, 10);
245+
}
246+
})
247+
.catch(err => {
248+
console.warn(`[${tag}] ${debug ? `(after ${commas(Date.now() - dryRunStart)} msecs) ` : " "}Dry-run failed: ${err}`);
249+
metrics.errors += 1;
250+
return null;
251+
});
252+
253+
// skip notarization if the dry-run failed
254+
if (currentValue === null) reject();
255+
256+
// compute and evaluate current deviation with respect to last notarized value
257+
const deviation = lastUpdates[caption].value > 0
258+
? (100 * (currentValue - lastUpdates[caption].value)) /
259+
lastUpdates[caption].value
260+
: 0;
261+
if (Math.abs(deviation) < conditions.deviationPercentage) {
262+
console.info(
263+
`[${tag}] ${deviation >= 0 ? "+" : ""}${deviation.toFixed(2)} % deviation after ${heartbeatSecs} secs.`
264+
)
265+
resolve(); // skip notarization
258266
} else {
259267
console.info(
260-
`[${tag}] Updating due to heartbeat after ${heartbeatSecs} secs ...`,
268+
`[${tag}] Updating due to price deviation of ${deviation.toFixed(2)} % ...`,
261269
);
262270
}
271+
} else {
272+
console.info(
273+
`[${tag}] Updating due to heartbeat after ${heartbeatSecs} secs ...`,
274+
);
275+
}
276+
277+
// create and sign and send new data request transaction
278+
console.debug(`[${tag}] Cache info before sending =>`, ledger.cacheInfo);
279+
const DRs = Witnet.DataRequests.from(ledger, request);
280+
metrics.inflight += 1;
281+
priceFeeds[caption].inflight = (priceFeeds[caption].inflight || 0) + 1;
282+
283+
// send data request transaction and wait for the notarization of a new price update
284+
await DRs.sendTransaction({
285+
fees: priority,
286+
witnesses: conditions.minWitnesses,
287+
}).then(tx => {
288+
console.info(`[${tag}] Sending data request transaction => { radHash: ${tx.radHash
289+
} inputs: ${tx.tx?.DataRequest?.signatures.length
290+
} cost: ${Witnet.Coins.fromNanowits(tx.fees.nanowits + tx.value?.nanowits).wits
291+
} weight: ${commas(tx.weight)
292+
} witnesses: ${tx.witnesses
293+
} hash: ${tx.hash
294+
} }`);
295+
metrics.nanowits += tx.fees.nanowits + tx.value?.nanowits;
296+
metrics.queries += 1;
297+
return DRs.confirmTransaction(tx.hash, {
298+
onStatusChange: () => console.info(`[${tag}] DRT status =>`, tx.status),
299+
})
300+
301+
}).then(async tx => {
302+
console.debug(
303+
`[${tag}] Cache info after confirmation =>`,
304+
ledger.cacheInfo,
305+
);
263306

264-
console.debug(`[${tag}] Cache info before sending =>`, ledger.cacheInfo);
265-
266-
// create, sign and send new data request transaction
267-
const DRs = Witnet.DataRequests.from(ledger, request);
268-
metrics.inflight += 1;
269-
priceFeeds[caption].inflight = (priceFeeds[caption].inflight || 0) + 1;
270-
271-
// launch promise for the notarization of new price update
272-
DRs.sendTransaction({
273-
fees: priority,
274-
witnesses: conditions.minWitnesses,
275-
}).then(tx => {
276-
console.info(`[${tag}] Sending data request transaction => { radHash: ${tx.radHash
277-
} inputs: ${tx.tx?.DataRequest?.signatures.length
278-
} cost: ${Witnet.Coins.fromNanowits(tx.fees.nanowits + tx.value?.nanowits).wits
279-
} weight: ${commas(tx.weight)
280-
} witnesses: ${tx.witnesses
281-
} hash: ${tx.hash
282-
} }`);
283-
metrics.nanowits += tx.fees.nanowits + tx.value?.nanowits;
284-
metrics.queries += 1;
285-
return DRs.confirmTransaction(tx.hash, {
286-
onStatusChange: () => console.info(`[${tag}] DRT status =>`, tx.status),
287-
})
288-
289-
}).then(async tx => {
290-
console.debug(
291-
`[${tag}] Cache info after confirmation =>`,
292-
ledger.cacheInfo,
307+
// await resolution in Witnet
308+
let status = tx.status;
309+
do {
310+
const report = await ledger.provider.getDataRequest(
311+
tx.hash,
312+
"ethereal",
293313
);
294-
295-
// await resolution in Witnet
296-
let status = tx.status;
297-
do {
298-
const report = await ledger.provider.getDataRequest(
299-
tx.hash,
300-
"ethereal",
314+
if (report.status !== status) {
315+
status = report.status;
316+
console.info(`[${tag}] DRT status =>`, report.status);
317+
}
318+
if (report.status === "solved" && report?.result) {
319+
const result = utils.cbor.decode(
320+
utils.fromHexString(report.result.cbor_bytes),
301321
);
302-
if (report.status !== status) {
303-
status = report.status;
304-
console.info(`[${tag}] DRT status =>`, report.status);
305-
}
306-
if (report.status === "solved" && report?.result) {
307-
const result = utils.cbor.decode(
308-
utils.fromHexString(report.result.cbor_bytes),
309-
);
310-
if (Number.isInteger(result)) {
311-
lastUpdates[caption].timestamp = report.result.timestamp;
312-
lastUpdates[caption].value = parseInt(result, 10);
313-
const { value, timestamp } = lastUpdates[caption]
314-
const providers = request.sources
315-
.map(source => {
316-
let parts = source.authority.split(".").slice(-2);
317-
parts[0] = parts[0][0].toUpperCase() + parts[0].slice(1);
318-
return parts.join(".")
319-
})
320-
.sort();
321-
console.info(`[${tag}] DRT result => { value: ${value}, ts: ${moment.unix(timestamp).format("MMM Do YYYY HH:mm:ss")}, providers: ${providers.join(" ")} }`);
322-
} else {
323-
throw `Unexpected DRT result => ${result}`;
324-
}
325-
break;
322+
if (Number.isInteger(result)) {
323+
lastUpdates[caption].timestamp = report.result.timestamp;
324+
lastUpdates[caption].value = parseInt(result, 10);
325+
const { value, timestamp } = lastUpdates[caption]
326+
const providers = request.sources
327+
.map(source => {
328+
let parts = source.authority.split(".").slice(-2);
329+
parts[0] = parts[0][0].toUpperCase() + parts[0].slice(1);
330+
return parts.join(".")
331+
})
332+
.sort();
333+
console.info(`[${tag}] DRT result => { value: ${value}, ts: ${moment.unix(timestamp).format("MMM Do YYYY HH:mm:ss")}, providers: ${providers.join(" ")} }`);
334+
} else {
335+
throw `Unexpected DRT result => ${result}`;
326336
}
327-
const delay = (ms) =>
328-
new Promise((_resolve) => setTimeout(_resolve, ms));
329-
await delay(5000);
330-
} while (status !== "solved");
331-
332-
}).then(() => {
333-
priceFeeds[caption].inflight -= 1;
334-
metrics.inflight -= 1;
335-
336-
}).catch(err => {
337-
priceFeeds[caption].inflight -= 1;
338-
metrics.inflight -= 1;
339-
metrics.errors += 1;
340-
console.error(`[${tag}] Notarization failed: ${err}`);
341-
});
342-
})
343-
.catch(err => {
344-
console.warn(`[${tag}] ${debug ? `(after ${commas(Date.now() - dryRunStart)} msecs) ` : " "}Dry-run failed: ${err}`);
345-
metrics.errors += 1;
346-
})
347-
);
348-
337+
break;
338+
}
339+
const delay = (ms) =>
340+
new Promise((_resolve) => setTimeout(_resolve, ms));
341+
await delay(5000);
342+
} while (status !== "solved");
343+
344+
}).then(() => {
345+
priceFeeds[caption].inflight -= 1;
346+
metrics.inflight -= 1;
347+
348+
}).catch(err => {
349+
priceFeeds[caption].inflight -= 1;
350+
metrics.inflight -= 1;
351+
metrics.errors += 1;
352+
console.error(`[${tag}] Notarization failed: ${err}`);
353+
});
354+
resolve();
355+
}));
356+
349357
if (threadBucket.length >= maxThreads) {
350358
await Promise.all(threadBucket)
351359
threadBucket = []
352360
}
353361
}
362+
354363
if (threadBucket.length) {
355364
await Promise.all(threadBucket)
356365
}

0 commit comments

Comments
 (0)