stronk/LivepeerEvents
1
0
Fork 0
mirror of https://github.com/stronk-dev/LivepeerEvents.git synced 2025-07-05 10:45:10 +02:00
Code Issues Packages Projects Releases Wiki Activity

Hopefully fix the infinite sync stopping after a few days. Else we go back to the WS push bases subscription

Browse Source 
Prework on displaying round numbers for events
This commit is contained in:
Marco van Dijk 2022-04-28 11:29:32 +02:00
parent d513bb9427
commit 9e71c63582
1 changed files with 72 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

143
backend/src/routes/livepeer.js
View File

@ -1412,76 +1412,80 @@ const initSync = async function () {
}); });
} }
let cycle = 0;
// Does the actual looping over last parsed block -> latest block in chain // Does the actual looping over last parsed block -> latest block in chain
const handleSync = async function () { const handleSync = async function () {
if (!CONF_DISABLE_DB && !startedInitSync) { if (!CONF_DISABLE_DB && !startedInitSync) {
console.log("Preloading all the things from the database"); console.log("Preloading all the things from the database");
await initSync(); await initSync();
} }
cycle++;
console.log('Starting new sync cycle #' + cycle);
isSyncing = true; isSyncing = true;
while (true) { // Get latest block in chain
// Get latest block in chain const latestBlock = await web3layer2.eth.getBlockNumber();
const latestBlock = await web3layer2.eth.getBlockNumber(); if (latestBlock > latestBlockInChain) {
if (latestBlock > latestBlockInChain) { latestBlockInChain = latestBlock;
latestBlockInChain = latestBlock; console.log("Latest L2 Eth block changed to " + latestBlockInChain);
console.log("Latest L2 Eth block changed to " + latestBlockInChain); } else {
} else { // If there are no new blocks, wait for 10 seconds before retrying
// If there are no new blocks, wait for 10 seconds before retrying console.log("No new blocks. Sleeping for 10 seconds...");
console.log("No new blocks. Sleeping for 10 seconds..."); setTimeout(() => {
await sleep(10000); handleSync();
continue; }, 10000)
} }
console.log("Needs to sync " + (latestBlockInChain - lastBlockEvents) + " blocks for Events sync"); console.log("Needs to sync " + (latestBlockInChain - lastBlockEvents) + " blocks for Events sync");
console.log("Needs to sync " + (latestBlockInChain - lastBlockTickets) + " blocks for Tickets sync"); console.log("Needs to sync " + (latestBlockInChain - lastBlockTickets) + " blocks for Tickets sync");
// Batch requests when sync is large, mark if we are going to reach latestBlockInChain in this round // Batch requests when sync is large, mark if we are going to reach latestBlockInChain in this round
let getFinalTickets = false; let getFinalTickets = false;
let toTickets = 'latest'; let toTickets = 'latest';
if (latestBlock - lastBlockTickets > 1000000) { if (latestBlock - lastBlockTickets > 1000000) {
toTickets = lastBlockTickets + 1000000; toTickets = lastBlockTickets + 1000000;
} else { } else {
getFinalTickets = true; getFinalTickets = true;
} }
let getFinalEvents = false; let getFinalEvents = false;
let toEvents = 'latest'; let toEvents = 'latest';
if (latestBlock - lastBlockEvents > 1000000) { if (latestBlock - lastBlockEvents > 1000000) {
toEvents = lastBlockEvents + 1000000; toEvents = lastBlockEvents + 1000000;
} else { } else {
getFinalEvents = true; getFinalEvents = true;
} }
// Start initial sync for this sync round // Start initial sync for this sync round
syncTickets(toTickets); syncTickets(toTickets);
syncEvents(toEvents); syncEvents(toEvents);
// Then loop until we have reached the last known block // Then loop until we have reached the last known block
while (isEventSyncing || isTicketSyncing || !getFinalTickets || !getFinalEvents) { while (isEventSyncing || isTicketSyncing || !getFinalTickets || !getFinalEvents) {
await sleep(500); await sleep(500);
if (isEventSyncing) { if (isEventSyncing) {
console.log("Parsed " + lastBlockEvents + " out of " + latestBlockInChain + " blocks for Event sync"); console.log("Parsed " + lastBlockEvents + " out of " + latestBlockInChain + " blocks for Event sync");
} else if (!getFinalEvents) { } else if (!getFinalEvents) {
// Start next batch for events // Start next batch for events
toEvents = 'latest'; toEvents = 'latest';
if (latestBlock - lastBlockEvents > 1000000) { if (latestBlock - lastBlockEvents > 1000000) {
toEvents = lastBlockEvents + 1000000; toEvents = lastBlockEvents + 1000000;
} else { } else {
getFinalEvents = true; getFinalEvents = true;
}
syncEvents(toEvents);
} }
if (isTicketSyncing) { syncEvents(toEvents);
console.log("Parsed " + lastBlockTickets + " out of " + latestBlockInChain + " blocks for Ticket sync"); }
} else if (!getFinalTickets) { if (isTicketSyncing) {
// Start next batch for tickets console.log("Parsed " + lastBlockTickets + " out of " + latestBlockInChain + " blocks for Ticket sync");
toTickets = 'latest'; } else if (!getFinalTickets) {
if (latestBlock - lastBlockTickets > 1000000) { // Start next batch for tickets
toTickets = lastBlockTickets + 1000000; toTickets = 'latest';
} else { if (latestBlock - lastBlockTickets > 1000000) {
getFinalTickets = true; toTickets = lastBlockTickets + 1000000;
} } else {
syncTickets(toTickets); getFinalTickets = true;
} }
syncTickets(toTickets);
} }
} }
console.log('done syncing')
isSyncing = false; isSyncing = false;
setTimeout(() => {
handleSync();
}, 10000)
}; };
if (!isSyncing && !CONF_SIMPLE_MODE) { if (!isSyncing && !CONF_SIMPLE_MODE) {
console.log("Starting sync process"); console.log("Starting sync process");
@ -2654,8 +2658,9 @@ Mutates the Event in the database to contain the round number
*/ */
let roundCache = []; let roundCache = [];
let highestEndblockSeen = 0;
const mutateRoundToDB = async function (scoreObj) { const mutateRound = async function (scoreObj) {
// Immediately mutate new object // Immediately mutate new object
const doc = await MonthlyStat.findOneAndUpdate({ const doc = await MonthlyStat.findOneAndUpdate({
year: year, year: year,
@ -2666,7 +2671,7 @@ const mutateRoundToDB = async function (scoreObj) {
upsert: true, upsert: true,
new: true, new: true,
setDefaultsOnInsert: true setDefaultsOnInsert: true
}); }); k
// Then find and mutate all Event objects which fall in this round // Then find and mutate all Event objects which fall in this round
} }
@ -2674,25 +2679,21 @@ apiRouter.post("/getRoundAtBlock", async (req, res) => {
try { try {
const { blockNumber } = req.body; const { blockNumber } = req.body;
if (blockNumber) { if (blockNumber) {
// Since months get counted starting at 0
const now = new Date().getTime();
let wasInCache = false;
// See if it is cached // See if it is cached
for (const thisAddr of orchScoreCache) { for (const thisRound of roundCache) {
if (thisAddr.year === year && thisAddr.month === month) { if (thisRound.startBlock <= blockNumber && thisRound.endBlock >= blockNumber) {
// Check timeout if (thisRound.endBlock > highestEndblockSeen) {
if (now - thisAddr.timestamp < 360000) {
res.send(thisAddr);
return;
} }
wasInCache = true; res.send(thisAddr);
} }
} }
// Check DB // Check DB
// If exists, mutate cache and return value
// Get from thegraph // Get from thegraph
console.log("Getting new Orchestrator scores for " + year + "-" + month + " @ " + url); console.log("Getting new Orchestrator scores for " + year + "-" + month + " @ " + url);
// Save to DB // Save to DB
mutateRoundToDB(); mutateRound();
} }
} catch (err) { } catch (err) {
console.log(err); console.log(err);