Make sequencer leader behavior optional on pds (#1250)

* make sequencer leader behavior optional on pds * tidy
bluesky-social · Jul 17, 2023 · 1ebda12 · 1ebda12
1 parent 670daf5
commit 1ebda12
Show file tree

Hide file tree

Showing 4 changed files with 25 additions and 19 deletions.
diff --git a/packages/dev-env/src/network.ts b/packages/dev-env/src/network.ts
@@ -62,6 +62,9 @@ export class TestNetwork extends TestNetworkNoAppView {
       await wait(50)
       if (!sub) return
       const state = await sub.getState()
+      if (!this.pds.ctx.sequencerLeader) {
+        throw new Error('Sequencer leader not configured on the pds')
+      }
       const caughtUp = await this.pds.ctx.sequencerLeader.isCaughtUp()
       if (!caughtUp) continue
       const { lastSeq } = await db

diff --git a/packages/pds/src/config.ts b/packages/pds/src/config.ts
@@ -58,6 +58,7 @@ export interface ServerConfigValues {
   maxSubscriptionBuffer: number
   repoBackfillLimitMs: number
   sequencerLeaderLockId?: number
+  sequencerLeaderEnabled?: boolean
 
   // this is really only used in test environments
   dbTxLockNonce?: string
@@ -192,6 +193,12 @@ export class ServerConfig {
       undefined,
     )
 
+    // by default each instance is a potential sequencer leader, but may be configured off
+    const sequencerLeaderEnabled = process.env.SEQUENCER_LEADER_ENABLED
+      ? process.env.SEQUENCER_LEADER_ENABLED !== '0' &&
+        process.env.SEQUENCER_LEADER_ENABLED !== 'false'
+      : undefined
+
     const dbTxLockNonce = nonemptyString(process.env.DB_TX_LOCK_NONCE)
 
     const bskyAppViewEndpoint = nonemptyString(
@@ -249,6 +256,7 @@ export class ServerConfig {
       maxSubscriptionBuffer,
       repoBackfillLimitMs,
       sequencerLeaderLockId,
+      sequencerLeaderEnabled,
       dbTxLockNonce,
       bskyAppViewEndpoint,
       bskyAppViewDid,
@@ -460,6 +468,10 @@ export class ServerConfig {
     return this.cfg.sequencerLeaderLockId
   }
 
+  get sequencerLeaderEnabled() {
+    return this.cfg.sequencerLeaderEnabled !== false
+  }
+
   get dbTxLockNonce() {
     return this.cfg.dbTxLockNonce
   }

diff --git a/packages/pds/src/context.ts b/packages/pds/src/context.ts
@@ -38,7 +38,7 @@ export class AppContext {
       services: Services
       messageDispatcher: MessageDispatcher
       sequencer: Sequencer
-      sequencerLeader: SequencerLeader
+      sequencerLeader: SequencerLeader | null
       labeler: Labeler
       labelCache: LabelCache
       backgroundQueue: BackgroundQueue
@@ -121,7 +121,7 @@ export class AppContext {
     return this.opts.sequencer
   }
 
-  get sequencerLeader(): SequencerLeader {
+  get sequencerLeader(): SequencerLeader | null {
     return this.opts.sequencerLeader
   }
 

diff --git a/packages/pds/src/index.ts b/packages/pds/src/index.ts
@@ -59,11 +59,7 @@ export class PDS {
   private dbStatsInterval?: NodeJS.Timer
   private sequencerStatsInterval?: NodeJS.Timer
 
-  constructor(opts: {
-    ctx: AppContext
-    app: express.Application
-    sequencerLeader: SequencerLeader
-  }) {
+  constructor(opts: { ctx: AppContext; app: express.Application }) {
     this.ctx = opts.ctx
     this.app = opts.app
   }
@@ -106,10 +102,9 @@ export class PDS {
 
     const messageDispatcher = new MessageDispatcher()
     const sequencer = new Sequencer(db)
-    const sequencerLeader = new SequencerLeader(
-      db,
-      config.sequencerLeaderLockId,
-    )
+    const sequencerLeader = config.sequencerLeaderEnabled
+      ? new SequencerLeader(db, config.sequencerLeaderLockId)
+      : null
 
     const mailTransport =
       config.emailSmtpUrl !== undefined
@@ -231,11 +226,7 @@ export class PDS {
     app.use(server.xrpc.router)
     app.use(error.handler)
 
-    return new PDS({
-      ctx,
-      app,
-      sequencerLeader,
-    })
+    return new PDS({ ctx, app })
   }
 
   async start(): Promise<http.Server> {
@@ -261,15 +252,15 @@ export class PDS {
       }, 10000)
     }
     this.sequencerStatsInterval = setInterval(() => {
-      if (this.ctx.sequencerLeader.isLeader) {
+      if (this.ctx.sequencerLeader?.isLeader) {
         seqLogger.info(
           { seq: this.ctx.sequencerLeader.peekSeqVal() },
           'sequencer leader stats',
         )
       }
     }, 500)
     appviewConsumers.listen(this.ctx)
-    this.ctx.sequencerLeader.run()
+    this.ctx.sequencerLeader?.run()
     await this.ctx.sequencer.start()
     await this.ctx.db.startListeningToChannels()
     this.ctx.labelCache.start()
@@ -283,7 +274,7 @@ export class PDS {
 
   async destroy(): Promise<void> {
     this.ctx.labelCache.stop()
-    await this.ctx.sequencerLeader.destroy()
+    await this.ctx.sequencerLeader?.destroy()
     await this.terminator?.terminate()
     await this.ctx.backgroundQueue.destroy()
     await this.ctx.db.close()