Cluster: A way to remove processed requests from storage

What is the problem this feature would solve?

Base: effect 4.1.1 (effect/cluster, SqlMessageStorage). main has the same code.

I looked through MessageStorage on main, the issues and the PRs, and found nothing for this, but I may have missed it.

A persisted request is deduplicated by its row in cluster_messages, and answered from cluster_replies. Nothing
ever deletes these rows once the request is processed. Two operations remove them, and neither fits a request that
is finished:

• clearReplies(requestId) deletes the reply, then sets processed = false on the request. The request is
  delivered again and the handler runs again.
• clearAddress(address) deletes every message and reply of the entity. That works only for an entity that is
  gone.

As far as I can tell, the tables of an entity that lives a long time grow with every request it handles, with no
limit. The workflow engine's rows grow the same way, apart from clearAddress on the clock entity. The application
often knows when a request can no longer be retried. For example, the sender's work has finished, so nobody will
send that primary key again. But I found no API to tell storage.

Repro (SingleRunner on @effect/sql-sqlite-bun, one entity whose handler counts its runs):

let runs = 0
const Hit = Rpc.make("Hit", { payload: { id: Schema.String }, success: Schema.Number, primaryKey: ({ id }) => id })
const Counter = Entity.make("Counter", [Hit])
  .annotateRpcs(ClusterSchema.Persisted, true)
  .annotateRpcs(ClusterSchema.WithTransaction, true)
const CounterLive = Counter.toLayer({ Hit: () => Effect.sync(() => ++runs) })

const client = (yield* Counter.client)("one")
yield* client.Hit({ id: "a" })
yield* client.Hit({ id: "a" })                       // deduplicated
for (let i = 0; i < 100; i++) yield* client.Hit({ id: `b${i}` })

yield* storage.clearReplies(requestIdOf("a"))         // nobody sends "a" again
yield* sharding.pollStorage

yield* sql`DELETE FROM cluster_replies WHERE request_id = ${requestIdOf("b0")}`
yield* sql`DELETE FROM cluster_messages WHERE request_id = ${requestIdOf("b0")}`
yield* client.Hit({ id: "b0" })

Step	Handler runs	Rows in cluster_messages / cluster_replies
"a" sent twice	1	1 / 1
100 more requests, then a wait	101	101 / 101
clearReplies on "a", then a poll	102: "a" ran again	101 / 101
rows of "b0" deleted by hand	102	100 / 100
"b0" sent again	103: a new request	101 / 101

Is there an intended way to remove processed requests? For example, are long-lived entities not the intended use,
with short-lived entities cleared by clearAddress instead? If so, a note in the docs would help, and the rest of
this issue can be ignored.

What is the feature you are proposing to solve the problem?

A way to forget processed requests, deleting the request and its replies together without
delivering it again. After that, a request sent with the same primary key would be new.

One possible shape (a sketch, open to a better one; I don't know the constraints you see):

// MessageStorage
readonly forgetRequests: (
  requestIds: ReadonlyArray<Snowflake.Snowflake>
) => Effect.Effect<void, PersistenceError>

• SqlMessageStorage deletes, in one transaction, every reply of each request, its chunks included, and every message of it: the request and its AckChunk and Interrupt envelopes. It works by request_id, so a message_id stored as its hash makes no difference. It does this only where the request is processed. An unprocessed request is left alone, so a message is never dropped before it is handled. Processed rows are no longer read by the mailbox (unprocessedMessages) or by resetShards, so the runner that owns the shard does not miss them.
• It runs inside the caller's transaction when there is one, like the other writes. So a WithTransaction
  handler can forget other requests in the same commit as its own writes.
• MemoryMessageStorage does the same.
• Once a fence like the one in #10058 exists, forgetRequests goes through it, like clearReplies and
  clearAddress.

The contract the caller would keep. A forgotten request leaves nothing for a waiting caller to read. I read the
first point below from the source, but could not reproduce it on SingleRunner, where a duplicate gets its reply
from memory:

• A caller that waits for the reply from storage never gets it. This is the read loop in Runners.ts, used when
  the request was sent from another runner, when its handler was dropped (for example, the shard moved), or for a
  duplicate. A runner knows only about the reply handlers registered on itself, not about callers on other runners,
  so storage cannot check this. The caller of forgetRequests must know that nobody waits for the request any more.
• The same primary key sent again is a new request, and the handler runs again.

What it would not cover:

• A handler's own request. A WithTransaction handler cannot forget its own request in its own commit. The
  reply is saved after the handler returns, so deleting the request inside the handler leaves an orphan row in
  cluster_replies. Forgetting it in afterCommit works, but a crash between the commit and the forget leaves
  the rows. An option on the request ("forget once replied") would close this. I have no design for it.
• Workflows. Forgetting a workflow's or an activity's request would make ClusterWorkflowEngine see it as never
  run, and run it again. A safe cleanup there would be an engine operation that purges a finished execution, not
  this one.

What alternatives have you considered?

• Short-lived entities cleared with clearAddress. This works when an entity's whole history can go at once.
  It does not fit an entity that lives a long time and handles many requests, where only the finished ones should
  go.
• clearReplies. It deletes the reply but delivers the request again, so the handler runs again.
• A time-based option (keep processed requests for a set duration). It needs no call from the application, but
  it is safe only where the application knows that no retry and no waiting caller comes after that time. So to me
  the explicit call seems the more general of the two, but you may see it differently.
• The application's own DELETE statements on Cluster's tables, as in the repro. That works, but it ties the
  application to Cluster's schema, which can change in any release, and I would rather use a supported way.