Repository navigation
ateapi: apply a replica's own worker writes to its cache at commit - #1583
Conversation
15879b0 to
3f55345
Compare
| // Version fencing deduplicates later watch events. Deletes are ignored | ||
| // to prevent out-of-order watch replays from resurrecting workers. | ||
| func (c *Cache) ApplyLocal(event store.WorkerEvent) { | ||
| if event.Type == store.WorkerEventDeleted { |
There was a problem hiding this comment.
I don't think we should ignore worker deletion? The worker name is the Pod UID, a worker resurrecting is not something that can realistically happen.
There was a problem hiding this comment.
got it, this makes sense.
even if the scheduler picks a non-existent worker, the bind would eventually fail.
| } | ||
| // Apply local writes directly to the cache on commit to eliminate outbox | ||
| // polling delay; later watch events are deduplicated by version. | ||
| if publisher, ok := persistence.(interface { |
There was a problem hiding this comment.
Should we do this registration in WorkerCache.Start?
| if err := tx.Commit(ctx); err != nil { | ||
| return nil, fmt.Errorf("committing transaction: %w", err) | ||
| } | ||
| if worker != nil && p.localEvents != nil && eventType != store.WorkerEventDeleted { |
There was a problem hiding this comment.
I think we should always publish events, including delete.
| if err != nil { | ||
| return err | ||
| } | ||
| if publisher, ok := c.store.(interface { |
There was a problem hiding this comment.
I feel this can be brittle. If the type assertion fails (e.g. the store gets wrapped into some sort of delegate pattern), the type assertion will silently fail and the performance optimization just gets disabled. I could be hard to troubleshoot.
Have you considered a model where we don't need the downstream consumers of the store API to register? The store itself could keep a registry of the watches it has registered, and publish WorkerEvent on them. This would need zero changes outside the store/atepg package.
| // worker events ahead of outbox delivery. Later outbox duplicates are deduplicated | ||
| // by version. | ||
| func (p *Persistence) PublishEventsLocally(sink func(store.WorkerEvent)) { | ||
| p.localEvents = sink |
There was a problem hiding this comment.
This can be brittle, if you have multiple callers registering sinks, only the last one will get the events. See my other comment about whether we can come up with a model where clients of the store don't need to register themselves.
There was a problem hiding this comment.
Originally I had registered the sink directly so the cache updated synchronously — ApplyEvent called inline at commit — since workercache is the only watcher.
Switched to the registry you suggested: Persistence keeps its own set of live WatchWorkers channels and publishes committed events onto them, so nothing downstream registers and there's no capability to probe for. workercache reads them off the same channel the outbox poller feeds. This trades away the synchronous guarantee.
Locally published events also arrive ahead of their xid position and again when the poll reaches them, so the channel now carries duplicates. Gap-free delivery is unaffected — the cursor only advances on the poller's send — and workercache version-fences creates and updates, so both are no-ops there.
The worker cache used to register a callback with the store, discovered through a type assertion. Wrapping the store in a delegate would have made that assertion fail silently, disabling the optimization with no signal. The store now keeps its own set of live WatchWorkers channels and publishes committed events onto them, so nothing downstream registers and there is no capability to probe for. Events are published from the payload the outbox stores rather than the in-memory Worker. The two differ: the stored encoding discards unknown fields, so the cache was previously handed unpruned messages on the fast path and pruned ones on the poll path for the same event. Locally published events reach a watcher ahead of their xid position and again when the poll delivers them, so WatchWorkers consumers must now tolerate duplicates as well as reconcile versions.
14f22f7 to
92fff2e
Compare
a30645f
into
agent-substrate:main
|
Thanks! |
Fixes #1370
Summary
atepgkeeps a registry of the liveWatchWorkerschannels and publishes each committed worker event onto them at commit, about one poll interval ahead of the outbox row carrying the same event. Replicas see their own writes sooner without a second notification API.Why
Replicas previously learned of their own writes only from the outbox watch, ~50ms later. In that window the scheduler re-picks a worker it just assigned and retries on the precondition check, and capacity a replica just freed stays invisible to its own placement. Cross-replica propagation is unchanged.
Key changes
WatchWorkersenrolls its channel; the poller deregisters before closing it, which is what makes a send on a closed channel impossible.writeAndAppendEventpublishes only aftertx.Commitreturns.workercachealready fences creates and updates on version, so duplicates and stale events are no-ops.Testing
TestLocalPublishReachesWatchers,TestLocalPublishSurvivesWatchClose, andTestCache_WatchEventsAreFencedare new.TestWatchWorkers_DeliveryFencedByOldestTransactionnow consumes the commit-time copy first and asserts the xmin fence on the outbox copy; the fence governs the poll, and a local publish bypasses it by construction.Full
atepgandworkercachesuites pass under-race.