Support auto scaling (#171)

* added heroku autoscaling plan

* prepare auto-scaling

* solved sonarcloud issues

* fixed tests

* allow storing logs to S3 bucket

* Revert "allow storing logs to S3 bucket"

This reverts commit a16dfc1.

* fixed log files upload to s3 in multi dyno environment

* add missing db migrations

* refactored job class

* feat: export JobRepository, RecordSocketAssociationRepository, ConnectedSocketRepository from package entry point

- Add exports for JobRepository, RecordSocketAssociationRepository, and ConnectedSocketRepository to the main package entry point
- Add type export for JobRow to support job-queue-persistence and record-concurrency dependent tasks

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* feat: make job.survey_id nullable so global jobs can be persisted

* solved potential issue in tests

* fixed issues reported by Copilot

* feat: bump minor version

---------

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
Co-authored-by: Stefano Ricci <SteRiccio@users.noreply.github.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

Author: 3 people

.env.template

@@ -29,6 +29,9 @@ LOG_RETENTION_DAYS=30
 LOG_UPLOAD_INTERVAL_MS=60000
 LOG_S3_PREFIX=logs
 ## Optional S3-backed log shipping (requires LOG_S3_ENABLED=true and file storage S3 to be enabled)
+## Log files are uploaded under a per-instance S3 prefix (LOG_S3_PREFIX/<instanceId>/...) to avoid
+## different instances overwriting each other's logs. The instance id is taken from DYNO (set
+## automatically by Heroku) or HOSTNAME, falling back to the OS hostname; no manual setup is usually needed.
 LOG_S3_ENABLED=false
 FILE_STORAGE_AWS_ACCESS_KEY=
 FILE_STORAGE_AWS_SECRET_ACCESS_KEY=

.yarn/install-state.gz

@@ -0,0 +1,109 @@
+import { DB } from '../db'
+import { Logger } from '../log'
+import { WsRelayMessageRepository } from '../repository'
+import { runWithClusterLock } from './clusterLock'
+import { ClusterEvent, ClusterEventHandler } from './types'
+
+type ListenConnection = Awaited<ReturnType<typeof DB.connect>>
+
+// Postgres caps NOTIFY payloads at 8000 bytes; stay safely under that before spilling to the DB.
+const NOTIFY_PAYLOAD_SAFE_THRESHOLD_BYTES = 7000
+
+const CHANNEL = 'arena_cluster_event'
+
+// Spilled-over payloads are only ever read moments after insert, by the NOTIFY that points at
+// them - a generous TTL here just bounds ws_relay_message's size if a dyno never claims one.
+const RELAY_MESSAGE_TTL_MS = 10 * 60_000
+const RELAY_MESSAGE_SWEEP_INTERVAL_MS = 5 * 60_000
+const RELAY_MESSAGE_SWEEP_LOCK_NAME = 'ws-relay-message-ttl-sweep'
+
+type InlineEnvelope = { inline: true; event: ClusterEvent }
+type RelayedEnvelope = { inline: false; relayMessageId: string }
+type Envelope = InlineEnvelope | RelayedEnvelope
+
+/**
+ * Postgres-backed cluster bus: a single LISTEN/NOTIFY channel shared by every dyno.
+ * `publish` never throws - delivery is best-effort, matching the pre-existing
+ * "check before emitting, self-heal otherwise" behavior of WebSocketServer.
+ */
+export class ClusterBus {
+  private static readonly logger: Logger = new Logger('ClusterBus')
+  private static listenConnection: ListenConnection | null = null
+  private static readonly handlers: ClusterEventHandler[] = []
+  private static sweepInterval: NodeJS.Timeout | null = null
+
+  static async init(): Promise<void> {
+    if (ClusterBus.listenConnection) return
+
+    const connection = await DB.connect({ direct: true })
+    ClusterBus.listenConnection = connection
+
+    connection.client.on('notification', (msg: { payload?: string }) => {
+      ClusterBus.onNotification(msg).catch((error) => ClusterBus.logger.error(`error handling notification: ${error}`))
+    })
+
+    await connection.none(`LISTEN ${CHANNEL}`)
+    ClusterBus.logger.info(`listening on channel "${CHANNEL}"`)
+
+    ClusterBus.sweepInterval = setInterval(() => {
+      runWithClusterLock({
+        lockName: RELAY_MESSAGE_SWEEP_LOCK_NAME,
+        fn: async () => {
+          await WsRelayMessageRepository.deleteExpired(RELAY_MESSAGE_TTL_MS)
+        },
+      }).catch((error) => ClusterBus.logger.error(`error running ws_relay_message TTL sweep: ${error}`))
+    }, RELAY_MESSAGE_SWEEP_INTERVAL_MS)
+    ClusterBus.sweepInterval.unref()
+  }
+
+  static async shutdown(): Promise<void> {
+    if (ClusterBus.sweepInterval) clearInterval(ClusterBus.sweepInterval)
+    ClusterBus.sweepInterval = null
+
+    const connection = ClusterBus.listenConnection
+    if (!connection) return
+
+    ClusterBus.listenConnection = null
+    try {
+      await connection.none(`UNLISTEN ${CHANNEL}`)
+    } catch (error) {
+      ClusterBus.logger.error(`error unlistening: ${error}`)
+    } finally {
+      await connection.done()
+    }
+  }
+
+  /**
+   * Registers a handler invoked for every cluster event received, including this dyno's own
+   * publications (Postgres delivers NOTIFY to every session listening on the channel).
+   * Handlers are expected to no-op when the event doesn't target something they own locally.
+   */
+  static onEvent(handler: ClusterEventHandler): void {
+    ClusterBus.handlers.push(handler)
+  }
+
+  static async publish(event: ClusterEvent): Promise<void> {
+    try {
+      const serializedEvent = JSON.stringify(event)
+
+      const envelope: Envelope =
+        Buffer.byteLength(serializedEvent, 'utf8') <= NOTIFY_PAYLOAD_SAFE_THRESHOLD_BYTES
+          ? { inline: true, event }
+          : { inline: false, relayMessageId: await WsRelayMessageRepository.insert(event) }
+
+      await DB.query('SELECT pg_notify($1, $2)', [CHANNEL, JSON.stringify(envelope)])
+    } catch (error) {
+      ClusterBus.logger.error(`error publishing event: ${error}`)
+    }
+  }
+
+  private static async onNotification(msg: { payload?: string }): Promise<void> {
+    if (!msg.payload) return
+
+    const envelope: Envelope = JSON.parse(msg.payload)
+    const event = envelope.inline ? envelope.event : await WsRelayMessageRepository.getById(envelope.relayMessageId)
+    if (!event) return
+
+    ClusterBus.handlers.forEach((handler) => handler(event as ClusterEvent))
+  }
+}

docs/superpowers/specs/2026-08-06-heroku-horizontal-autoscaling-design.md

@@ -0,0 +1,33 @@
+import { DB } from '../db'
+import { Logger } from '../log'
+
+const logger: Logger = new Logger('ClusterLock')
+
+/**
+ * Runs `fn` while holding a cluster-wide Postgres advisory lock named `lockName`.
+ * Non-blocking: if another dyno already holds the lock, `fn` is skipped and `false` is returned.
+ * The lock is session-scoped, acquired and released on the same pooled connection via `DB.task`,
+ * so it is always released even if `fn` throws.
+ *
+ * @param params - Lock name and the function to run while holding it
+ */
+export const runWithClusterLock = async (params: { lockName: string; fn: () => Promise<void> }): Promise<boolean> => {
+  const { lockName, fn } = params
+
+  return DB.task(async (task) => {
+    const { locked } = await task.one<{ locked: boolean }>('SELECT pg_try_advisory_lock(hashtext($1)) AS locked', [
+      lockName,
+    ])
+    if (!locked) return false
+
+    try {
+      await fn()
+      return true
+    } catch (error) {
+      logger.error(`error running task under cluster lock "${lockName}": ${error}`)
+      throw error
+    } finally {
+      await task.one('SELECT pg_advisory_unlock(hashtext($1))', [lockName])
+    }
+  })
+}

src/clusterBus/clusterBus.ts

@@ -0,0 +1,3 @@
+export { ClusterBus } from './clusterBus'
+export { runWithClusterLock } from './clusterLock'
+export type { ClusterEvent, ClusterEventHandler } from './types'

src/clusterBus/clusterLock.ts

@@ -0,0 +1,86 @@
+import { DB } from '../../db'
+import { DBMigrator } from '../../db/dbMigrator'
+import { ClusterBus } from '../clusterBus'
+import { runWithClusterLock } from '../clusterLock'
+import { ClusterEvent } from '../types'
+
+let received: ClusterEvent[] = []
+
+const waitFor = async (predicate: () => boolean, timeoutMs = 2000): Promise<void> => {
+  const start = Date.now()
+  while (!predicate()) {
+    if (Date.now() - start > timeoutMs) throw new Error('timed out waiting for cluster event')
+    await new Promise((resolve) => setTimeout(resolve, 20))
+  }
+}
+
+// Top-level (not per-describe) hooks: ClusterBus/DB are file-wide singletons, so init and
+// pool teardown must each run exactly once for the whole file, regardless of how many
+// describe blocks use them.
+beforeAll(async () => {
+  await DBMigrator.migrateSchema()
+  await ClusterBus.init()
+  // A single handler collecting every event: ClusterBus.onEvent has no matching "off", so
+  // registering per-test would leak handlers across tests within this file.
+  ClusterBus.onEvent((event) => received.push(event))
+})
+
+afterEach(() => {
+  received = []
+})
+
+afterAll(async () => {
+  await ClusterBus.shutdown()
+  await DB.$pool.end()
+})
+
+describe('ClusterBus', () => {
+  test('publish delivers an inline event back to this dyno through its own listener', async () => {
+    await ClusterBus.publish({ targetType: 'test', targetId: 'inline-1', eventType: 'ping', message: { n: 1 } })
+
+    await waitFor(() => received.length === 1)
+    expect(received[0]).toEqual({ targetType: 'test', targetId: 'inline-1', eventType: 'ping', message: { n: 1 } })
+  })
+
+  test('publish spills oversized payloads through ws_relay_message and still delivers them', async () => {
+    const bigMessage = { blob: 'x'.repeat(8000) }
+
+    await ClusterBus.publish({ targetType: 'test', targetId: 'big-1', eventType: 'ping-big', message: bigMessage })
+
+    await waitFor(() => received.length === 1)
+    expect(received[0].message).toEqual(bigMessage)
+  })
+
+  test('publish never throws, even if the event cannot be delivered', async () => {
+    const circular: Record<string, unknown> = {}
+    circular.self = circular // not JSON-serializable
+
+    await expect(
+      ClusterBus.publish({ targetType: 'test', targetId: 'bad-1', eventType: 'ping', message: circular })
+    ).resolves.toBeUndefined()
+  })
+})
+
+describe('runWithClusterLock', () => {
+  test('only one of two concurrent callers runs, and the lock is released afterwards', async () => {
+    const lockName = `test-lock-${Date.now()}`
+    let runs = 0
+    const runFn = async (): Promise<void> => {
+      runs++
+      await new Promise((resolve) => setTimeout(resolve, 200))
+    }
+
+    const [a, b] = await Promise.all([
+      runWithClusterLock({ lockName, fn: runFn }),
+      runWithClusterLock({ lockName, fn: runFn }),
+    ])
+
+    expect([a, b].filter(Boolean)).toHaveLength(1)
+    expect(runs).toBe(1)
+
+    // the lock was released after the first run completed, so a later caller can acquire it
+    const c = await runWithClusterLock({ lockName, fn: runFn })
+    expect(c).toBe(true)
+    expect(runs).toBe(2)
+  })
+})

src/clusterBus/index.ts

@@ -0,0 +1,13 @@
+/**
+ * Generic "invalidate/deliver this on every dyno" envelope. `targetType`/`targetId` are opaque
+ * to ClusterBus - each subsystem (WebSocketServer, record/survey cache invalidation, ...)
+ * defines its own vocabulary and is expected to no-op on events it doesn't recognize.
+ */
+export interface ClusterEvent {
+  targetType: string
+  targetId: string
+  eventType: string
+  message: any
+}
+
+export type ClusterEventHandler = (event: ClusterEvent) => void

src/clusterBus/tests/clusterBus.test.ts

@@ -0,0 +1,51 @@
+'use strict'
+
+var dbm
+var type
+var seed
+var fs = require('fs')
+var path = require('path')
+var Promise
+
+/**
+ * We receive the dbmigrate dependency from dbmigrate initially.
+ * This enables us to not have to rely on NODE_PATH.
+ */
+exports.setup = function (options, seedLink) {
+  dbm = options.dbmigrate
+  type = dbm.dataType
+  seed = seedLink
+  Promise = options.Promise
+}
+
+exports.up = function (db) {
+  var filePath = path.join(__dirname, 'sqls', '20260806120000-add-tables-cluster-bus-up.sql')
+  return new Promise(function (resolve, reject) {
+    fs.readFile(filePath, { encoding: 'utf-8' }, function (err, data) {
+      if (err) return reject(err)
+      console.log('received data: ' + data)
+
+      resolve(data)
+    })
+  }).then(function (data) {
+    return db.runSql(data)
+  })
+}
+
+exports.down = function (db) {
+  var filePath = path.join(__dirname, 'sqls', '20260806120000-add-tables-cluster-bus-down.sql')
+  return new Promise(function (resolve, reject) {
+    fs.readFile(filePath, { encoding: 'utf-8' }, function (err, data) {
+      if (err) return reject(err)
+      console.log('received data: ' + data)
+
+      resolve(data)
+    })
+  }).then(function (data) {
+    return db.runSql(data)
+  })
+}
+
+exports._meta = {
+  version: 1,
+}

src/clusterBus/types.ts

@@ -0,0 +1,51 @@
+'use strict'
+
+var dbm
+var type
+var seed
+var fs = require('fs')
+var path = require('path')
+var Promise
+
+/**
+ * We receive the dbmigrate dependency from dbmigrate initially.
+ * This enables us to not have to rely on NODE_PATH.
+ */
+exports.setup = function (options, seedLink) {
+  dbm = options.dbmigrate
+  type = dbm.dataType
+  seed = seedLink
+  Promise = options.Promise
+}
+
+exports.up = function (db) {
+  var filePath = path.join(__dirname, 'sqls', '20260819100000-create-table-job-up.sql')
+  return new Promise(function (resolve, reject) {
+    fs.readFile(filePath, { encoding: 'utf-8' }, function (err, data) {
+      if (err) return reject(err)
+      console.log('received data: ' + data)
+
+      resolve(data)
+    })
+  }).then(function (data) {
+    return db.runSql(data)
+  })
+}
+
+exports.down = function (db) {
+  var filePath = path.join(__dirname, 'sqls', '20260819100000-create-table-job-down.sql')
+  return new Promise(function (resolve, reject) {
+    fs.readFile(filePath, { encoding: 'utf-8' }, function (err, data) {
+      if (err) return reject(err)
+      console.log('received data: ' + data)
+
+      resolve(data)
+    })
+  }).then(function (data) {
+    return db.runSql(data)
+  })
+}
+
+exports._meta = {
+  version: 1,
+}