Cluster: processedRequestIds grows without bound with MessageStorage.layerNoop

[capy-ai]

What version of Effect is running?

effect 4.0.0-rc.116 (effect/unstable/cluster), Bun 1.3

What steps can reproduce the bug?

Run a Sharding with MessageStorage.layerNoop and send requests to entities that then hibernate:

import { heapStats } from "bun:jsc"
import { Effect, Layer, Schema } from "effect"
import { Entity, MessageStorage, RunnerHealth, Runners, RunnerStorage, Sharding, ShardingConfig } from "effect/unstable/cluster"
import { Rpc } from "effect/unstable/rpc"

const Ping = Entity.make("Ping", [Rpc.make("Ping", { success: Schema.Number })])
const PingLive = Ping.toLayer(Effect.succeed({ Ping: () => Effect.succeed(1) }), { maxIdleTime: "250 millis" })

const ShardingLive = Sharding.layer.pipe(
  Layer.provideMerge(Runners.layerNoop),
  Layer.provideMerge(MessageStorage.layerNoop),
  Layer.provide([RunnerStorage.layerMemory, RunnerHealth.layerNoop]),
  Layer.provide(ShardingConfig.layer({ shardsPerGroup: 1 })),
)

const live = () => {
  Bun.gc(true)
  const s = heapStats()
  return { bytes: s.heapSize - s.extraMemorySize, objects: s.objectCount }
}

const N = 100_000

await Effect.runPromise(
  Effect.gen(function* () {
    const client = yield* Ping.client
    const send = (n: number) =>
      Effect.forEach(Array.from({ length: n }, (_, i) => i), (i) => client(`e-${i % 10}`).Ping(), { concurrency: 16, discard: true })
    yield* send(10_000)
    yield* Effect.sleep("12 seconds")
    const before = live()
    yield* send(N)
    yield* Effect.sleep("12 seconds") // every entity has been reaped
    const after = live()
    console.log(`+${((after.bytes - before.bytes) / N).toFixed(1)} B, +${((after.objects - before.objects) / N).toFixed(2)} objects per request`)
  }).pipe(Effect.provide(PingLive.pipe(Layer.provideMerge(ShardingLive)))),
)

What is the expected behavior?

Once every entity has hibernated, retained heap stays flat no matter how many requests were processed.

What do you see instead?

+93.6 B, +1.00 objects per request, and it never drops. With the processedRequestIds.add(...) line below removed, the same run gives +6.4 B, -0.01 objects.

Additional information

The cause: EntityManager adds every replied request id to processedRequestIds (internal/entityManager.ts, around lines 147 and 288). Only clearProcessed() clears that set, and only the storage read loop calls it (Sharding.ts:774). That loop starts only when storage !== MessageStorage.noop (Sharding.ts:240, 625), so with layerNoop the set grows for the runner's whole lifetime. That's about 94 B per request, roughly 1 GiB per 10 million requests.

One possible fix is to skip the add when storage is disabled, since a request that isn't persisted can't be redelivered from storage. Another is to clear the set when an entity is removed.