Hyperlinkv0.8.0-beta.28

Polling

Polling.adaptiveconstsrc/Polling.ts:611
(options: {
  readonly active: Duration.Input
  readonly idle: Duration.Input
  readonly factor?: number
}): Layer.Layer<PollingTag>

Work-aware cadence — the complement of backoff. Ticks run at active after a work signal and DECAY toward idle (each tick multiplies the wait by factor, capped at idle) while nothing happens. Signal work by calling resetCadence — from inside the effect via current, from the handle via proc.polling.resetCadence, or wire a stream to it with wakeOn. The reset snaps the cadence back to active and wakes the current wait.

The queue-drainer shape: drain fast while entries keep arriving, back off to a lazy idle poll when the queue runs dry.

Source src/Polling.ts:61145 lines
export const adaptive = (options: {
  readonly active: Duration.Input;
  readonly idle: Duration.Input;
  readonly factor?: number;
}): Layer.Layer<PollingTag> => {
  const activeMs = Duration.toMillis(Duration.fromInputUnsafe(options.active));
  const idleMs = Duration.toMillis(Duration.fromInputUnsafe(options.idle));
  const factor = options.factor ?? 2;
  const delayMs = (iteration: number): number =>
    Math.min(activeMs * Math.pow(factor, iteration), idleMs);

  return registerPollingLayer(
    Layer.effect(
      PollingTag,
      Effect.gen(function* () {
        const iteration = yield* Ref.make(0);
        const wakeRef = yield* Ref.make<Deferred.Deferred<void, never>>(
          Deferred.makeUnsafe()
        );
        const requestWake = Effect.flatMap(Ref.get(wakeRef), (d) =>
          Deferred.succeed(d, undefined)
        ).pipe(Effect.asVoid);
        const awaitNextTick = Effect.gen(function* () {
          const d = Deferred.makeUnsafe<void, never>();
          yield* Ref.set(wakeRef, d);
          const n = yield* Ref.get(iteration);
          yield* Effect.race(
            Effect.sleep(Duration.millis(delayMs(n))),
            Deferred.await(d)
          ).pipe(Effect.asVoid);
        });
        return {
          awaitNextTick,
          requestWake,
          // work signal: snap back to `active` and end the current wait
          resetCadence: Ref.set(iteration, 0).pipe(Effect.andThen(requestWake)),
          afterTick: Ref.update(iteration, (n) => n + 1),
          peekCadence: Effect.map(Ref.get(iteration), (n) =>
            Option.some(Duration.millis(delayMs(n)))
          ),
        } satisfies PollingService;
      })
    )
  );
};