Meters fed by a stream

Electricity meters fed by a stream of their readings, as a partition of Kafka gives them. The boundary reads each reading from the source, sends it to its meter under a request id derived from its offset, and acknowledges it only once the meter stored its answer, Accepted or Rejected. A meter answered Suspended or Full is sent the reading again, a meter answered NotFound is created at its first reading, and a reading answered Reused stops the feed. The meter keeps the offset of the last reading applied, and rejects a reading the source gives again, so a crash before an acknowledgement counts nothing twice, after the reply's keep too. A first version of the meter dies on a meter replaced, and the feed waits at that reading until a fix applies it.

• meter.ts: the meter.
• feed.ts: the boundary between the source and the runtime.
• feed.test.ts: the feed on the runtime, with a source that gives its readings again until each is acknowledged.

feed.ts

/**
 * The feed's boundary: what the application runs to read a stream of meter readings into the meters. The source is
 * a log of offsets, as a partition of Kafka is, which gives again every reading after the last one acknowledged. A
 * reading is acknowledged only once the meter stored its answer, `Accepted` or `Rejected`: any other answer stored
 * nothing of the reading, and an acknowledgement would lose it (decision 0272).
 */
import * as Data from 'effect/Data'
import * as Effect from 'effect/Effect'
import type * as Schedule from 'effect/Schedule'
import type { Handle, TypesOf } from '@fgaudo/effect-tea-persistent'
import type { meter } from './meter.ts'

/** A reading of the source: its offset in the log, the meter it is of, and the meter's index. */
export type Reading = Readonly<{ offset: number; meter: string; index: number }>

/** The source of the readings, which gives each one again until it is acknowledged. */
export interface Source {
  /** The readings after the last one acknowledged, in the order of their offsets. */
  readonly unacknowledged: Effect.Effect<ReadonlyArray<Reading>>
  readonly acknowledge: (reading: Reading) => Effect.Effect<void>
}

/** A reading whose meter stored nothing of it yet: it is suspended, or holds as many messages as its bound. */
export class Unanswered extends Data.TaggedError('Unanswered')<{ reading: Reading; answer: 'Suspended' | 'Full' }> {}

/**
 * A reading answered `Reused`: its request id was given before to another message. The source gave another reading
 * under an offset, or a deploy changed how a reading is written. Only an operator tells them apart.
 */
export class ReusedOffset extends Data.TaggedError('ReusedOffset')<{ reading: Reading }> {}

export type Options = Readonly<{
  /** How a reading whose meter stored nothing of it is sent again. */
  retry: Schedule.Schedule<unknown, Unanswered | ReusedOffset>
  /** How long a reply is kept. The meter rejects a reading sent again after it, so the keep only spares a step. */
  keep: Readonly<{ millis: number }>
}>

const feed = { _tag: 'Feed' } as const

/**
 * Sends a reading to its meter until the meter stores its answer. It ends once the reading may be acknowledged, and
 * fails without acknowledging otherwise.
 */
export const deliver = (meters: Handle<TypesOf<typeof meter>>, options: Options) => (reading: Reading) => {
  const message = { _tag: 'Read', offset: reading.offset, index: reading.index } as const
  // The request id is derived from the offset, so a reading the source gives again is the same request.
  const sent = { requestId: `reading-${reading.offset}`, sender: feed, keep: options.keep }
  const attempt = Effect.gen(function*() {
    let answer = yield* meters.send(reading.meter, message, sent)
    // A meter exists from its first reading: the feed knows every meter the source names.
    if (answer._tag === 'NotFound') {
      yield* meters.create(reading.meter, { requestId: `meter-${reading.meter}`, sender: feed })
      answer = yield* meters.send(reading.meter, message, sent)
    }
    switch (answer._tag) {
      // A stale reading was applied already, and is acknowledged as one applied.
      case 'Accepted':
      case 'Rejected':
        return
      case 'Suspended':
      case 'Full':
        return yield* new Unanswered({ reading, answer: answer._tag })
      case 'Reused':
        return yield* new ReusedOffset({ reading })
      case 'NotFound':
        return yield* Effect.die(`the meter ${reading.meter} was created and is not found`)
    }
  })
  return Effect.retry(attempt, { schedule: options.retry, while: (error) => error._tag === 'Unanswered' })
}

/**
 * Feeds the meters with the readings the source gives, one at a time and in order, so that a meter is given its
 * readings in the order of their offsets. A reading whose meter stays suspended stops the feed there, and the
 * readings after it wait for it, those of other meters too.
 */
export const run = (meters: Handle<TypesOf<typeof meter>>, source: Source, options: Options) =>
  Effect.flatMap(source.unacknowledged, (readings) =>
    Effect.forEach(
      readings,
      (reading) => Effect.andThen(deliver(meters, options)(reading), source.acknowledge(reading)),
      { discard: true },
    ))
Copy

meter.ts

/**
 * An electricity meter, fed by a stream of its readings. A reading gives the meter's index, the kilowatt-hours it
 * counted since it was installed, and the meter's consumption grows by the difference. The state keeps the offset
 * of the last reading applied, so a reading the source gives again is rejected, whatever the reply's keep.
 */
import * as Schema from 'effect/Schema'
import { program, rejectWith } from '@fgaudo/effect-tea-persistent/program'

const Count = Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))

const State = Schema.Struct({
  // The offset of the last reading applied, or null before the first one.
  last: Schema.NullOr(Count),
  index: Count,
  kwh: Count,
})

const Inbound = Schema.TaggedStruct('Read', { offset: Count, index: Count })

// A reading at or before the last offset applied was applied already.
const Rejection = Schema.Literals(['Stale'])

/** The feed's boundary is the only sender. */
const Sender = Schema.TaggedStruct('Feed', {})

const Projection = Schema.Struct({ last: Schema.NullOr(Schema.Int), kwh: Schema.Int })

export const meter = program({
  state: State,
  flags: Schema.Null,
  inbound: Inbound,
  rejection: Rejection,
  sender: Sender,
  projection: Projection,
})({
  name: 'Meter',

  // A meter is installed at an index of 0.
  init: () => ({ last: null, index: 0, kwh: 0 }),

  update: ({ offset, index }) => (state) =>
    state.last !== null && offset <= state.last
      ? rejectWith('Stale')
      // A meter's index never goes back, so a lower index is a meter replaced, which counts again from 0.
      : { last: offset, index, kwh: state.kwh + (index >= state.index ? index - state.index : index) },

  project: ({ last, kwh }) => ({ last, kwh }),
})
Copy

feed.test.ts

/**
 * The feed on the runtime: a source that gives its readings again until each is acknowledged, and meters that store
 * each reading once. A reading is acknowledged only once its meter stored its answer (decision 0272).
 */
import { SqliteClient } from '@effect/sql-sqlite-bun'
import { expect, test } from 'bun:test'
import * as Effect from 'effect/Effect'
import * as Ref from 'effect/Ref'
import * as Schedule from 'effect/Schedule'
import type * as Scope from 'effect/Scope'
import type { SqlClient } from 'effect/sql'
import { launch } from '@fgaudo/effect-tea-persistent'
import { defect } from '@fgaudo/effect-tea-persistent/program'
import { type Options, type Reading, run, type Source } from './feed.ts'
import { meter } from './meter.ts'

const database = <A, E>(test: Effect.Effect<A, E, SqlClient.SqlClient | Scope.Scope>) =>
  Effect.runPromise(
    Effect.scoped(test).pipe(
      Effect.provide(SqliteClient.layer({ filename: ':memory:' })),
      Effect.orDie,
    ),
  )

/** A log of readings, which gives every reading after the last one acknowledged, as a partition of Kafka does. */
const log = (readings: ReadonlyArray<Reading>) =>
  Effect.map(Ref.make(-1), (acknowledged) => {
    const source: Source = {
      unacknowledged: Effect.map(Ref.get(acknowledged), (last) => readings.filter(({ offset }) => offset > last)),
      acknowledge: ({ offset }) => Ref.set(acknowledged, offset),
    }
    return { source, acknowledged: Ref.get(acknowledged), lost: Ref.set(acknowledged, -1) }
  })

const options: Options = { retry: Schedule.recurs(2), keep: { millis: 60_000 } }

// The meter as one would write it at first: a meter's index never goes back, so a lower index is a defect. A meter
// replaced counts again from 0, and suspends this version.
const first: typeof meter = {
  ...meter,
  update: (message, meta) => (state) =>
    message.index < state.index ? defect('a meter never goes back') : meter.update(message, meta)(state),
}

test('readings the source gives again are counted once, after the keep too', () =>
  database(Effect.gen(function*() {
    const runtime = yield* launch()
    const meters = yield* runtime.deploy(meter)
    const { source, acknowledged, lost } = yield* log([
      { offset: 0, meter: 'm-1', index: 10 },
      { offset: 1, meter: 'm-2', index: 4 },
      { offset: 2, meter: 'm-1', index: 25 },
    ])
    // A keep of no time at all: a reading given again is a new request, which the meter rejects as stale.
    const forgetful = { ...options, keep: { millis: 0 } }
    yield* run(meters, source, forgetful)
    expect(yield* acknowledged).toBe(2)
    // The feed created each meter at its first reading.
    expect(yield* meters.count({})).toBe(2)
    // The acknowledgements were lost, as by a crash before the source stored them: the source gives every reading again.
    yield* lost
    yield* run(meters, source, forgetful)
    yield* runtime.settled
    expect(yield* acknowledged).toBe(2)
    expect(yield* meters.read('m-1')).toEqual({ last: 2, kwh: 25 })
    expect(yield* meters.read('m-2')).toEqual({ last: 1, kwh: 4 })
  })))

const replaced: ReadonlyArray<Reading> = [
  { offset: 0, meter: 'm-1', index: 10 },
  { offset: 1, meter: 'm-1', index: 25 },
  // The meter was replaced: its index starts again from 0.
  { offset: 2, meter: 'm-1', index: 3 },
  { offset: 3, meter: 'm-1', index: 8 },
]

test('a reading its meter dies on is not acknowledged, and the fix applies it once', () =>
  database(Effect.gen(function*() {
    const { source, acknowledged } = yield* log(replaced)

    // The first process runs the first version, which dies on the replaced meter's reading.
    yield* Effect.scoped(Effect.gen(function*() {
      const runtime = yield* launch()
      const meters = yield* runtime.deploy(first)
      // The meter is answered `Suspended` at each try: it stored nothing of the reading yet.
      expect(yield* Effect.flip(run(meters, source, options))).toMatchObject({
        _tag: 'Unanswered',
        answer: 'Suspended',
        reading: { offset: 2 },
      })
      expect(yield* acknowledged).toBe(1)
      expect((yield* meters.suspended).map(({ id }) => id)).toEqual(['m-1'])
    }))

    // The fix is a deploy, by a new process on the same database. The resume applies the reading the meter died on,
    // and the feed, given it again, is answered from the stored reply.
    yield* Effect.scoped(Effect.gen(function*() {
      const runtime = yield* launch()
      const meters = yield* runtime.deploy(meter)
      expect(yield* meters.resume('m-1')).toEqual({ _tag: 'Resumed' })
      yield* run(meters, source, options)
      yield* runtime.settled
      expect(yield* acknowledged).toBe(3)
      expect(yield* meters.read('m-1')).toEqual({ last: 3, kwh: 10 + 15 + 3 + 5 })
    }))
  })))

// A discard stores no reply for a caller's message, so the feed's next try is a new request, and the meter dies on it
// again. To get past the reading, the feed skips it, or a fix rejects it (decision 0272).
test('a discard alone does not get the feed past a reading its meter dies on', () =>
  database(Effect.gen(function*() {
    const { source, acknowledged } = yield* log(replaced)
    const runtime = yield* launch()
    const meters = yield* runtime.deploy(first)
    yield* Effect.flip(run(meters, source, options))
    expect(yield* meters.discard('m-1')).toEqual({ _tag: 'Discarded' })
    yield* runtime.settled
    expect(yield* Effect.flip(run(meters, source, options))).toMatchObject({
      _tag: 'Unanswered',
      answer: 'Suspended',
      reading: { offset: 2 },
    })
    expect(yield* acknowledged).toBe(1)
  })))

// The source gave another reading under an offset it gave before. The feed stops there, and an operator decides
// which reading is right.
test('a reading under an offset given before to another reading stops the feed, unacknowledged', () =>
  database(Effect.gen(function*() {
    const runtime = yield* launch()
    const meters = yield* runtime.deploy(meter)
    const before = yield* log([{ offset: 0, meter: 'm-1', index: 10 }])
    yield* run(meters, before.source, options)
    const after = yield* log([{ offset: 0, meter: 'm-1', index: 12 }])
    expect(yield* Effect.flip(run(meters, after.source, options))).toMatchObject({
      _tag: 'ReusedOffset',
      reading: { offset: 0 },
    })
    expect(yield* after.acknowledged).toBe(-1)
    expect(yield* meters.read('m-1')).toEqual({ last: 0, kwh: 10 })
  })))
Copy