ateapi: apply a replica's own worker writes to its cache at commit - #1583
ateapi: apply a replica's own worker writes to its cache at commit#1583shrutiyam-glitch wants to merge 3 commits into
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.
2 participants
Fixes #1370
Summary
Adds an optional in-process sink (
PublishEventsLocally) to synchronously update the local worker cache (workercache.ApplyLocal) at commit time. This provides "read-your-own-writes" consistency, bypassing the outbox polling delay. Later outbox duplicates are safely discarded by the cache's version fencing.Why
Previously, replicas learned of their own worker writes via the outbox watch (a ~50ms delay). This blind spot caused:
This change eliminates these self-inflicted conflicts. Cross-replica propagation remains unchanged (journal-backed).
Key changes
writeAndAppendEvent, the single funnel for all worker mutations.Testing
Added
TestPublishEventsLocally(Postgres sync firing, cloning, and local publishing).Added
TestCache_ApplyLocal_*tests (immediate application and version fencing).Full
atepgandworkercachesuites pass under-race.Tests pass
Appropriate changes to documentation are included in the PR