feat(cu): support using postgres as db #588

servers/cu/.postgres/Dockerfile

@@ -0,0 +1,6 @@
+# Only used for development purposes
+FROM postgres:17
+
+ENV POSTGRES_PASSWORD=admin
+ENV POSTGRES_USER=admin
+ENV POSTGRES_DB=cu

servers/cu/package.json

@@ -17,32 +17,33 @@
     "@permaweb/weavedrive": "^0.0.18",
     "arweave": "^1.15.5",
     "async-lock": "^1.4.1",
-    "better-sqlite3": "^11.5.0",
+    "better-sqlite3": "^11.7.0",
     "bytes": "^3.1.2",
     "cors": "^2.8.5",
-    "dataloader": "^2.2.2",
-    "dotenv": "^16.4.5",
+    "dataloader": "^2.2.3",
+    "dotenv": "^16.4.7",
     "fast-glob": "^3.3.2",
-    "fastify": "^5.1.0",
+    "fastify": "^5.2.0",
     "helmet": "^8.0.0",
     "hyper-async": "^1.1.2",
     "keccak": "^3.0.4",
     "long-timeout": "^0.1.1",
     "lru-cache": "^11.0.2",
     "ms": "^2.1.3",
-    "opossum": "^8.3.0",
-    "p-map": "^7.0.2",
+    "opossum": "^8.4.0",
+    "p-map": "^7.0.3",
     "p-queue": "^8.0.1",
+    "pg": "^8.13.1",
     "prom-client": "^15.1.3",
     "ramda": "^0.30.1",
-    "undici": "^6.20.1",
+    "undici": "^7.2.0",
     "warp-arbundles": "^1.0.4",
-    "winston": "^3.16.0",
+    "winston": "^3.17.0",
     "workerpool": "^9.2.0",
-    "zod": "^3.23.8"
+    "zod": "^3.24.1"
   },
   "devDependencies": {
-    "nodemon": "^3.1.7"
+    "nodemon": "^3.1.9"
   },
   "engines": {
     "node": ">=18"

servers/cu/src/bootstrap.js

@@ -14,8 +14,8 @@ import { fromPromise } from 'hyper-async'
 import lt from 'long-timeout'
 
 // Precanned clients to use for OOTB apis
+import * as DbClient from './effects/db.js'
 import * as ArweaveClient from './effects/arweave.js'
-import * as SqliteClient from './effects/sqlite.js'
 import * as AoSuClient from './effects/ao-su.js'
 import * as WasmClient from './effects/wasm.js'
 import * as AoProcessClient from './effects/ao-process.js'
@@ -77,8 +77,7 @@ export const createApis = async (ctx) => {
     cacheKeyFn: ({ processId }) => processId
   })
 
-  const DB_URL = `${ctx.DB_URL}.sqlite`
-  const sqlite = await SqliteClient.createSqliteClient({ url: DB_URL, bootstrap: true })
+  const db = await DbClient.createDbClient({ url: ctx.DB_URL, bootstrap: true })
 
   const BROADCAST = 'workers'
   const workerBroadcast = new BroadcastChannel(BROADCAST).unref()
@@ -98,7 +97,7 @@ export const createApis = async (ctx) => {
           ARWEAVE_URL: ctx.ARWEAVE_URL,
           GRAPHQL_URL: ctx.GRAPHQL_URL,
           CHECKPOINT_GRAPHQL_URL: ctx.CHECKPOINT_GRAPHQL_URL,
-          DB_URL,
+          DB_URL: ctx.DB_URL,
           id: workerId,
           MODE: ctx.MODE,
           LOG_CONFIG_PATH: ctx.LOG_CONFIG_PATH,
@@ -219,9 +218,9 @@ export const createApis = async (ctx) => {
     hashWasmMemory: WasmClient.hashWasmMemoryWith({ logger: ctx.logger }),
     buildAndSignDataItem: ArweaveClient.buildAndSignDataItemWith({ WALLET: ctx.WALLET }),
     uploadDataItem: ArweaveClient.uploadDataItemWith({ UPLOADER_URL: ctx.UPLOADER_URL, fetch: ctx.fetch, logger: ctx.logger }),
-    writeCheckpointRecord: AoProcessClient.writeCheckpointRecordWith({ db: sqlite }),
+    writeCheckpointRecord: AoProcessClient.writeCheckpointRecordWith({ db }),
     writeFileCheckpointMemory,
-    writeFileCheckpointRecord: AoProcessClient.writeFileCheckpointRecordWith({ db: sqlite }),
+    writeFileCheckpointRecord: AoProcessClient.writeFileCheckpointRecordWith({ db }),
     logger: ctx.logger,
     DISABLE_PROCESS_CHECKPOINT_CREATION: ctx.DISABLE_PROCESS_CHECKPOINT_CREATION,
     DISABLE_PROCESS_FILE_CHECKPOINT_CREATION: ctx.DISABLE_PROCESS_FILE_CHECKPOINT_CREATION,
@@ -307,14 +306,14 @@ export const createApis = async (ctx) => {
     loadTransactionMeta: ArweaveClient.loadTransactionMetaWith({ fetch: ctx.fetch, GRAPHQL_URL: ctx.GRAPHQL_URL, logger }),
     loadTransactionData: ArweaveClient.loadTransactionDataWith({ fetch: ctx.fetch, ARWEAVE_URL: ctx.ARWEAVE_URL, logger }),
     isProcessOwnerSupported: AoProcessClient.isProcessOwnerSupportedWith({ ALLOW_OWNERS: ctx.ALLOW_OWNERS }),
-    findProcess: AoProcessClient.findProcessWith({ db: sqlite, logger }),
+    findProcess: AoProcessClient.findProcessWith({ db, logger }),
     findLatestProcessMemory: AoProcessClient.findLatestProcessMemoryWith({
       cache: wasmMemoryCache,
       loadTransactionData: ArweaveClient.loadTransactionDataWith({ fetch: ctx.fetch, ARWEAVE_URL: ctx.ARWEAVE_URL, logger }),
       readProcessMemoryFile,
       readFileCheckpointMemory,
-      findFileCheckpointBefore: AoProcessClient.findFileCheckpointBeforeWith({ db: sqlite }),
-      findRecordCheckpointBefore: AoProcessClient.findRecordCheckpointBeforeWith({ db: sqlite }),
+      findFileCheckpointBefore: AoProcessClient.findFileCheckpointBeforeWith({ db }),
+      findRecordCheckpointBefore: AoProcessClient.findRecordCheckpointBeforeWith({ db }),
       address,
       queryGateway: ArweaveClient.queryGatewayWith({ fetch: ctx.fetch, GRAPHQL_URL: ctx.GRAPHQL_URL, logger }),
       queryCheckpointGateway: ArweaveClient.queryGatewayWith({ fetch: ctx.fetch, GRAPHQL_URL: ctx.CHECKPOINT_GRAPHQL_URL, logger }),
@@ -331,14 +330,14 @@ export const createApis = async (ctx) => {
     }),
     evaluationCounter,
     // gasCounter,
-    saveProcess: AoProcessClient.saveProcessWith({ db: sqlite, logger }),
-    findEvaluation: AoEvaluationClient.findEvaluationWith({ db: sqlite, logger }),
-    saveEvaluation: AoEvaluationClient.saveEvaluationWith({ db: sqlite, logger }),
-    findBlocks: AoBlockClient.findBlocksWith({ db: sqlite, logger }),
-    saveBlocks: AoBlockClient.saveBlocksWith({ db: sqlite, logger }),
+    saveProcess: AoProcessClient.saveProcessWith({ db, logger }),
+    findEvaluation: AoEvaluationClient.findEvaluationWith({ db, logger }),
+    saveEvaluation: AoEvaluationClient.saveEvaluationWith({ db, logger }),
+    findBlocks: AoBlockClient.findBlocksWith({ db, logger }),
+    saveBlocks: AoBlockClient.saveBlocksWith({ db, logger }),
     loadBlocksMeta: AoBlockClient.loadBlocksMetaWith({ fetch: ctx.fetch, GRAPHQL_URL: ctx.GRAPHQL_URL, pageSize: 90, logger }),
-    findModule: AoModuleClient.findModuleWith({ db: sqlite, logger }),
-    saveModule: AoModuleClient.saveModuleWith({ db: sqlite, logger }),
+    findModule: AoModuleClient.findModuleWith({ db, logger }),
+    saveModule: AoModuleClient.saveModuleWith({ db, logger }),
     loadEvaluator: AoModuleClient.evaluatorWith({
       loadWasmModule,
       evaluateWith: (prep) => primaryWorkQueue.add(() =>
@@ -362,7 +361,7 @@ export const createApis = async (ctx) => {
       ),
       logger
     }),
-    findMessageBefore: AoEvaluationClient.findMessageBeforeWith({ db: sqlite, logger }),
+    findMessageBefore: AoEvaluationClient.findMessageBeforeWith({ db, logger }),
     loadTimestamp: AoSuClient.loadTimestampWith({ fetch: ctx.fetch, logger }),
     loadProcess: AoSuClient.loadProcessWith({ fetch: ctx.fetch, logger }),
     loadMessages: AoSuClient.loadMessagesWith({ fetch: ctx.fetch, pageSize: 1000, logger }),
@@ -450,13 +449,13 @@ export const createApis = async (ctx) => {
   const readCronResultsLogger = ctx.logger.child('readCronResults')
   const readCronResults = readCronResultsWith({
     ...sharedDeps(readCronResultsLogger),
-    findEvaluations: AoEvaluationClient.findEvaluationsWith({ db: sqlite, logger: readCronResultsLogger })
+    findEvaluations: AoEvaluationClient.findEvaluationsWith({ db, logger: readCronResultsLogger })
   })
 
   const readResultsLogger = ctx.logger.child('readResults')
   const readResults = readResultsWith({
     ...sharedDeps(readResultsLogger),
-    findEvaluations: AoEvaluationClient.findEvaluationsWith({ db: sqlite, logger: readResultsLogger })
+    findEvaluations: AoEvaluationClient.findEvaluationsWith({ db, logger: readResultsLogger })
   })
 
   let checkpointP

servers/cu/src/config.js

@@ -117,7 +117,6 @@ const CONFIG_ENVS = {
     PROCESS_CHECKPOINT_CREATION_THROTTLE: process.env.PROCESS_CHECKPOINT_CREATION_THROTTLE || ms('30m'),
     DISABLE_PROCESS_CHECKPOINT_CREATION: process.env.DISABLE_PROCESS_CHECKPOINT_CREATION !== 'false',
     DISABLE_PROCESS_FILE_CHECKPOINT_CREATION: process.env.DISABLE_PROCESS_FILE_CHECKPOINT_CREATION !== 'false',
-    EAGER_CHECKPOINT_THRESHOLD: process.env.EAGER_CHECKPOINT_THRESHOLD || 100,
     /**
      *  EAGER_CHECKPOINT_ACCUMULATED_GAS_THRESHOLD: Amount of gas for 2 hours of continuous compute (300_000_000_000_000)
      *  This was calculated by creating a process built to do continuous compute. After 2 hours, this process used
@@ -167,7 +166,6 @@ const CONFIG_ENVS = {
     PROCESS_CHECKPOINT_CREATION_THROTTLE: process.env.PROCESS_CHECKPOINT_CREATION_THROTTLE || ms('30m'),
     DISABLE_PROCESS_CHECKPOINT_CREATION: process.env.DISABLE_PROCESS_CHECKPOINT_CREATION !== 'false', // TODO: disabled by default for now. Enable by default later
     DISABLE_PROCESS_FILE_CHECKPOINT_CREATION: process.env.DISABLE_PROCESS_FILE_CHECKPOINT_CREATION !== 'false',
-    EAGER_CHECKPOINT_THRESHOLD: process.env.EAGER_CHECKPOINT_THRESHOLD || 100,
     /**
      *  EAGER_CHECKPOINT_ACCUMULATED_GAS_THRESHOLD: Amount of gas for 2 hours of continuous compute (300_000_000_000_000)
      *  This was calculated by creating a process built to do continuous compute by adding and clearing a table.