Queue replication files in a sorted set to process

.env.sample

@@ -0,0 +1,4 @@
+AWS_ACCESS_KEY_ID=
+AWS_SECRET_ACCESS_KEY=
+OsmchaAdminToken=
+RedisServer='redis://redis:6379/0'

.gitignore

@@ -1,3 +1,4 @@
 node_modules/
 *.log.*
-.DS_Store
+.DS_Store
+.env

docker-compose.yaml

@@ -0,0 +1,15 @@
+version: '3.2'
+
+services:
+  redis:
+    image: redis:alpine
+    ports:
+      - "6379:6379"
+
+  shell:
+    build: .
+    volumes:
+      - .:/app
+    env_file:
+      - .env
+    command: sh -c "while true; do sleep 1000; done"

index.js

@@ -19,7 +19,7 @@ process.on('exit', (code) => {
 });
 
 const run = async (replicationFileId) => {
-  console.time('TIME');
+  console.time(`TIME ${replicationFileId}`);
   const { Body } = await s3.getObject({
     Bucket: REPLICATION_BUCKET,
     Key: formatReplicationKey(replicationFileId)
@@ -34,7 +34,7 @@ const run = async (replicationFileId) => {
 
   await Promise.all(changesetList.map(postTagChanges));
   console.log('Posted all changesets tagChanges to OSMCha API');
-  console.timeEnd('TIME');
+  console.timeEnd(`TIME ${replicationFileId}`);
 };
 
 module.exports = run;

update.js

@@ -1,23 +1,33 @@
 'use strict';
 
 const { getPlanetTimestamp } = require('./util/get-states');
-const { getLastProcessedState, setProcessedState } = require('./util/redis-client');
+const { getLastProcessedState, setProcessedState, storePendingReplications, getReplicationToProcess } = require('./util/redis-client');
 const { range } = require('./util/range');
 const run = require('./index');
 
 const process = async () => {
+  // check if there are any files to process and queue them
   const planetState = await getPlanetTimestamp();
 
   let lastProcessedState = await getLastProcessedState();
 
   if (!lastProcessedState) lastProcessedState = planetState - 2;
   const files = range(lastProcessedState, planetState + 1);
-
-  for (const f of files) {
-    await run(f);
-    await setProcessedState(f);
+  console.log(`Queueing replication files from ${lastProcessedState} to ${planetState} to process.`);
+
+  await Promise.all(files.map(async (f) => {
+    await storePendingReplications(f);
+  }));
+
+  // Now check the queue and pick a file to process
+  let toProcess = await getReplicationToProcess();
+  while (toProcess) {
+    console.log(`Processing replication file ${toProcess}`);
+    await run(toProcess);
+    console.log(`Finished processing replication file ${toProcess}`);
+    await setProcessedState(toProcess);
+    toProcess = await getReplicationToProcess();
   }
-  console.log(`Processed from ${lastProcessedState} to ${planetState}.`)
 };
 
 process();

util/redis-client.js

@@ -12,6 +12,26 @@ const setProcessedState = async (value) => {
   await client.disconnect();
 };
 
+const storePendingReplications = async (value) => {
+  const client = await createClient(REDIS_CONFIG)
+    .on('error', err => console.log('Redis Client Error', err))
+    .connect();
+
+  await client.zAdd('adiff-service:pending_replications_sorted', {value: `${value}`, score: value});
+  await client.disconnect();
+};
+
+const getReplicationToProcess = async () => {
+  const client = await createClient(REDIS_CONFIG)
+    .on('error', err => console.log('Redis Client Error', err))
+    .connect();
+
+  const replication = await client.zPopMin('adiff-service:pending_replications_sorted');
+  await client.disconnect();
+  return replication ? Number(replication.value) : null;
+};
+
+
 const getLastProcessedState = async () => {
   const client = await createClient(REDIS_CONFIG)
     .on('error', err => console.log('Redis Client Error', err))
@@ -24,5 +44,7 @@ const getLastProcessedState = async () => {
 
 module.exports = {
   setProcessedState,
+  storePendingReplications,
+  getReplicationToProcess,
   getLastProcessedState,
 };