fixup! Eliminate Redis race condition with atomic updates

Author: Jon-edge

src/indexEngines.ts

@@ -4,12 +4,26 @@ import { coinrankEngine } from './coinrankEngine'
 import { config } from './config'
 import { ratesEngine } from './ratesEngine'
 import { uidEngine } from './uidEngine'
+import {
+  setupCurrencyCodeMapsSync,
+  syncedCurrencyCodeMaps
+} from './utils/currencyCodeMapsSync'
 import { ratesDbSetup } from './utils/dbUtils'
 import { logger } from './utils/utils'
 
-// Initialize DB
+// Initialize DB and currency code maps sync
 async function initDb(): Promise<void> {
-  await setupDatabase(config.couchUri, ratesDbSetup)
+  // Create a new ratesDbSetup that includes the syncedCurrencyCodeMaps from the new sync module
+  const indexEnginesRatesDbSetup = {
+    ...ratesDbSetup,
+    syncedDocuments: [syncedCurrencyCodeMaps]
+  }
+
+  await setupDatabase(config.couchUri, indexEnginesRatesDbSetup)
+
+  // Set up currency code maps sync (only runs in this process)
+  setupCurrencyCodeMapsSync()
+
   ratesEngine().catch(e => logger('ratesEngine failure', e))
   uidEngine().catch(e => logger('uidEngine failure', e))
   coinrankEngine().catch(e => logger('coinrankEngine failure', e))

src/utils/currencyCodeMapsSync.ts

@@ -0,0 +1,62 @@
+import { asArray, asMaybe, asObject, asString } from 'cleaners'
+import { syncedDocument } from 'edge-server-tools'
+
+import currencyCodeMaps from './currencyCodeMaps.json'
+import { hsetAsync } from './dbUtils'
+import { logger } from './utils'
+
+const asCurrencyCodeMapsCleaner = asObject({
+  constantCurrencyCodes: asMaybe(asObject(asString)),
+  zeroRates: asMaybe(asObject(asString)),
+  fallbackConstantRates: asMaybe(asObject(asString)),
+  coinMarketCap: asMaybe(asObject(asString)),
+  coincap: asMaybe(asObject(asString)),
+  coingecko: asMaybe(asObject(asString)),
+  allEdgeCurrencies: asMaybe(asArray(asString)),
+  fiatCurrencyCodes: asMaybe(asArray(asString))
+})
+
+// Pass the defaults json through the cleaner so they're typed
+const defaultCurrencyCodeMaps = asCurrencyCodeMapsCleaner(currencyCodeMaps)
+
+const asCurrencyCodeMaps = asMaybe(
+  asCurrencyCodeMapsCleaner,
+  defaultCurrencyCodeMaps
+)
+
+export const syncedCurrencyCodeMaps = syncedDocument(
+  'currencyCodeMaps',
+  asCurrencyCodeMaps
+)
+
+// Set up the sync logic - this will only run in the indexEngines process
+export const setupCurrencyCodeMapsSync = (): void => {
+  syncedCurrencyCodeMaps.onChange(currencyCodeMaps => {
+    const timestamp = new Date().toISOString()
+    logger(
+      `[${timestamp}] SYNC TRIGGERED: Syncing currency code maps with redis cache...`
+    )
+    logger(
+      `[${timestamp}] SYNC TRIGGER: onChange fired for currencyCodeMaps document (PID: ${process.pid})`
+    )
+
+    for (const key of Object.keys(currencyCodeMaps)) {
+      const updateKey = async (): Promise<void> => {
+        try {
+          // Use hset directly as suggested by peachbits - simpler and avoids race conditions
+          if (Array.isArray(currencyCodeMaps[key])) {
+            await hsetAsync(key, Object.assign({}, currencyCodeMaps[key]))
+          } else {
+            await hsetAsync(key, currencyCodeMaps[key])
+          }
+          logger(`Successfully updated Redis key: ${key}`)
+        } catch (e) {
+          logger('currencyCodeMaps sync failed to update', key, e)
+        }
+      }
+
+      // Fire and forget - don't await in the callback
+      updateKey().catch(e => logger('Unhandled error in updateKey', key, e))
+    }
+  })
+}

src/utils/dbUtils.ts

@@ -1,12 +1,4 @@
-import {
-  asArray,
-  asBoolean,
-  asMaybe,
-  asObject,
-  asOptional,
-  asString
-} from 'cleaners'
-import { syncedDocument } from 'edge-server-tools'
+import { asBoolean, asMaybe, asObject, asOptional, asString } from 'cleaners'
 import nano from 'nano'
 import promisify from 'promisify-node'
 import { createClient } from 'redis'
@@ -153,80 +145,11 @@ export const getFromDb = async (
 export const wrappedGetFromDb = async (dates: string[]): Promise<DbDoc[]> =>
   getFromDb(dbRates, dates)
 
-const asCurrencyCodeMapsCleaner = asObject({
-  constantCurrencyCodes: asMaybe(asObject(asString)),
-  zeroRates: asMaybe(asObject(asString)),
-  fallbackConstantRates: asMaybe(asObject(asString)),
-  coinMarketCap: asMaybe(asObject(asString)),
-  coincap: asMaybe(asObject(asString)),
-  coingecko: asMaybe(asObject(asString)),
-  allEdgeCurrencies: asMaybe(asArray(asString)),
-  fiatCurrencyCodes: asMaybe(asArray(asString))
-})
-
-// Pass the defaults json through the cleaner so they're typed
-const defaultCurrencyCodeMaps = asCurrencyCodeMapsCleaner(currencyCodeMaps)
-
-const asCurrencyCodeMaps = asMaybe(
-  asCurrencyCodeMapsCleaner,
-  defaultCurrencyCodeMaps
-)
-
-const syncedCurrencyCodeMaps = syncedDocument(
-  'currencyCodeMaps',
-  asCurrencyCodeMaps
-)
-
-// Only run sync in background engine processes, not web server instances
-if (process.env.ENABLE_BACKGROUND_SYNC !== 'false') {
-  syncedCurrencyCodeMaps.onChange(currencyCodeMaps => {
-    const timestamp = new Date().toISOString()
-    logger(
-      `[${timestamp}] SYNC TRIGGERED: Syncing currency code maps with redis cache...`
-    )
-    logger(
-      `[${timestamp}] SYNC TRIGGER: onChange fired for currencyCodeMaps document (PID: ${process.pid})`
-    )
-    for (const key of Object.keys(currencyCodeMaps)) {
-      const tempKey = `${key}_temp_${Date.now()}`
-
-      // Write to temporary key first, then atomically rename
-      const writeToTemp = async (): Promise<void> => {
-        if (Array.isArray(currencyCodeMaps[key])) {
-          await hsetAsync(tempKey, Object.assign({}, currencyCodeMaps[key]))
-        } else {
-          await hsetAsync(tempKey, currencyCodeMaps[key])
-        }
-      }
-
-      const updateKey = async (): Promise<void> => {
-        try {
-          await writeToTemp()
-          // Atomically replace the old key with the new data
-          await renameAsync(tempKey, key)
-          logger(`Successfully updated Redis key: ${key}`)
-        } catch (e) {
-          logger('syncedCurrencyCodeMaps failed to update', key, e)
-          // Clean up temporary key on failure
-          try {
-            await delAsync(tempKey)
-          } catch (cleanupError) {
-            logger('Failed to cleanup temp key', tempKey, cleanupError)
-          }
-        }
-      }
-
-      // Fire and forget - don't await in the callback
-      updateKey().catch(e => logger('Unhandled error in updateKey', key, e))
-    }
-  })
-}
-
 export const ratesDbSetup = {
   name: 'db_rates',
   options: { partitioned: false },
   templates: { currencyCodeMaps },
-  syncedDocuments: [syncedCurrencyCodeMaps]
+  syncedDocuments: [] // Empty array since sync is now handled in indexEngines
 }
 
 export const getEdgeAssetDoc = memoize(