Balancer core

Usage sample

import {Db, migrateDb, MessageBalancer, createMethodPerLevelLoggerAdapter, MessageCache, MessageStorage} from '@targetprocess/balancer-core'
import {Pool} from 'pg'
import {makeLogger} from 'loggerism'
import * as promClient from 'prom-client'

type MessageBalancerA = MessageBalancer<{data?: string; retry?: number}>

type MessageBalancerB = MessageBalancer<{data?: string}>

async function main() {
  const [balancerA, balancerB] = await createAndInitBalancers()

  balancerA.storeMessage({
    partitionKey: 'partition#1',
    content: Buffer.alloc(128),
    properties: {data: 'some arbitrary data'}
  })

  balancerB.storeMessage({
    partitionKey: 'partition#2',
    content: Buffer.alloc(128),
    properties: {data: 'some arbitrary data'}
  })

  balancerA.processNextMessage(async message => {
    try {
      const {partitionGroup, partitionKey} = message
      console.log(`Processed message from partition "${partitionGroup}/${partitionKey}"`)
      return {type: 'Ok'}
    } catch {
      const properties = {
        ...message.properties,
        retry: (message.properties?.retry || 0) + 1
      }
      // Push message back with updated properties
      return {type: 'Requeue', update: {properties}}
    }
  })

  balancerB.processNextMessage(async message => {
    try {
      const {partitionGroup, partitionKey} = message
      console.log(`Processed message from partition "${partitionGroup}/${partitionKey}"`)
      return {type: 'Ok'}
    } catch {
      // Push message back with no properties update
      return {type: 'Requeue'}
    }
  })
}

async function createAndInitBalancers(): Promise<[MessageBalancerA, MessageBalancerB]> {
  const pool = new Pool({
    connectionString: process.env.POSTGRES_CONNECTION_STRING,
    max: process.env.POSTGRES_POOL_MAX
  })

  pool.on('error', error => {
    // Handle error here
    console.error(error)
  })

  await migrateDb({pool})

  const logger = createMethodPerLevelLoggerAdapter(makeLogger({
    logLevel: process.env.LOG_LEVEL,
    handleExceptions: false
  }))

  const db = new Db({pool})
  const storage = new MessageStorage({
    db,
    logger,
    createMessagesDurationMetric: summaryMetric('create_messages_duration_in_ms'),
    updateMessageDurationMetric: summaryMetric('update_message_duration_in_ms'),
    removeMessagesDurationMetric: summaryMetric('remove_messages_duration_in_ms'),
    readMessagesDurationMetric: summaryMetric('read_messages_duration_in_ms'),
    readMessagesOrderedByIdDurationMetric: summaryMetric('read_messages_ordered_by_id_duration_in_ms')
  })
  const cache = new MessageCache({
    maxMessageSize: Number(process.env.MESSAGE_CACHE_MAX_MESSAGE_SIZE),
    maxSize: Number(process.env.MESSAGE_CACHE_MAX_SIZE),
    logger,
    messageCountMetric: gaugeMetric('cache_message_count'),
    messageSizeMetric: gaugeMetric('cache_message_size')
  })
  const balancerA = new MessageBalancer<{data?: string; retry?: number}>({
    partitionGroup: 'A',
    partitionSizeLimit: Number(process.env.PARTITION_SIZE_LIMIT),
    lockPartition: true,
    storage,
    cache,
    logger,
    endToEndMessageProcessingDurationMetric: summaryMetric('balancer_a_end_to_end_processing_duration_in_ms'),
    centrifugePartitionCountMetric: gaugeMetric('balancer_a_centrifuge_partition_count'),
    centrifugeMessageCountMetric: gaugeMetric('balancer_a_centrifuge_message_count')
  })
  const balancerB = new MessageBalancer<{data?: string}>({
    partitionGroup: 'B',
    partitionSizeLimit: Number(process.env.PARTITION_SIZE_LIMIT),
    lockPartition: true,
    storage,
    cache,
    logger,
    endToEndMessageProcessingDurationMetric: summaryMetric('balancer_b_end_to_end_processing_duration_in_ms'),
    centrifugePartitionCountMetric: gaugeMetric('balancer_b_centrifuge_partition_count'),
    centrifugeMessageCountMetric: gaugeMetric('balancer_b_centrifuge_message_count')
  })

  await balancerA.init()
  await balancerB.init()

  return [
    balancerA,
    balancerA
  ]
}

function summaryMetric(name: string) {
  return new promClient.Summary({
    name,
    help: 'Write it yourself',
    percentiles: [0.1, 0.5, 0.9, 0.99],
    maxAgeSeconds: 10 * 60,
    ageBuckets: 10
  })
}

function gaugeMetric(name: string) {
  return new promClient.Gauge({
    name,
    help: 'Write it yourself'
  })
}

main()

The 0.6.x line is maintained separately

There are two independent lines of this package:

| line | where | used by | |---|---|---| | 0.6.x | support/0.6.0, branched from the 0.6.1 tag | the balancer service | | 1.x | master | the balancer-3 service |

They share a package name but have diverged far enough to be treated as separate products. The usage sample above documents 0.6.x and does not apply to 1.x.

support/0.6.0 is the long-lived branch for the whole 0.6 series, not for the 0.6.0 release. It is never merged into master.

Why they are kept apart

Because the balancer service is not meant to move onto the current line at all. It is meant to be replaced by the balancer-3 service, which already runs on it. Upgrading the dependency would put the effort into the wrong thing, so 0.6.x is kept alive only to support the old service until it is retired.

That holds for two reasons.

It would not be cheap. The 1.x API is not source-compatible with 0.6.x, so the service cannot cross over by changing a version range — it would have to be partly rewritten. The effort is comparable to adopting balancer-3 instead.

It would not fix what motivates moving. On 1.x a process that dies still loses its record of what it had already handed downstream, so those messages can be delivered a second time after a restart. balancer-3 keeps that record durably and restores it at startup, so the problem disappears rather than merely getting smaller.

What keeps the old service alive for now is that the replacement is more than a redeployment: balancer-3 expects the consuming service to confirm completion back to it, so that service needs work too. Until that is done, this line stays a holding position rather than a destination.