feat: add WithFilter and WithWatchPipeline options to CachedTable - #116
Conversation
CachedTable.InitializeWithConfig now accepts two new functional options: - WithFilter(filter any): scopes the initial eager load (FindMany) and ReconcilerGetAllKeys to only return entries matching the filter. The filter is passed directly to StoreCollection.FindMany. - WithWatchPipeline(pipeline any): scopes the change stream (Watch) to only deliver events matching the pipeline. The pipeline is passed directly to StoreCollection.Watch (expects a mongo.Pipeline). Both options are fully backward-compatible — existing callers that pass no options continue to get the current behavior (nil filter, nil pipeline = all entries, all events). This enables downstream consumers like GenericIntegrationTable to scope their cache loading and change stream to a specific subset of documents (e.g., filtering by integration type), preventing cross-tenant cache pollution and unnecessary reconciler triggers. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
WalkthroughAdded exported Changes
Sequence Diagram(s)sequenceDiagram
participant Initializer as Initializer
participant Collection as Collection (DB)
participant Cache as CachedTable
participant Reconciler as Reconciler
rect rgba(200,200,255,0.5)
Initializer->>Collection: col.Count(ctx, filter) -- validate filter
Collection-->>Initializer: count / error
end
rect rgba(200,255,200,0.5)
Initializer->>Collection: FindMany(ctx, filter) -- eager key enumeration
Collection-->>Initializer: keys / error
Initializer->>Cache: store keys
end
rect rgba(255,200,200,0.5)
Initializer->>Collection: col.Watch(ctx, watchPipeline, handler) -- register change stream
Collection-->>Cache: change events
Cache->>Reconciler: ReconcilerGetAllKeys() uses FindMany(ctx, filter)
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@table/cached_generic.go`:
- Around line 208-214: Replace the panic paths that call log.Panicf in
cached_generic.go with error returns and a single upfront filter preflight: when
eager-loading keys (the block using t.readThrough and t.col.FindMany) return the
encountered error to the caller (propagate it out of InitializeWithConfig)
instead of panicking; similarly change the other panic in the
ReconcilerGetAllKeys path to return an error. Add a preflight check of the
caller-supplied filter (WithFilter) during InitializeWithConfig by invoking
t.col.FindMany (or a lightweight validation call) once and returning any error
so read-through mode cannot defer a filter failure to ReconcilerGetAllKeys.
Update InitializeWithConfig and ReconcilerGetAllKeys signatures/returns as
needed to propagate these errors.
- Around line 162-173: After applying opts into CachedTableConfig, validate that
Filter and WatchPipeline are set together: if exactly one of config.Filter or
config.WatchPipeline is non-nil (XOR), reject the partial configuration (return
an error from the construction function or panic consistently with surrounding
API) and surface a clear message referencing WithFilter/WithWatchPipeline
pairing; update the code path that applies opts (the loop over opts and the
assignment to t.readThrough, t.filter, t.watchPipeline) to perform this check
before assigning to t.filter/t.watchPipeline, and update callers/tests
accordingly (or alternatively collapse them into a single scope option if you
prefer that design).
- Around line 96-99: The watch-pipeline example (the pipeline variable and its
use with InitializeWithConfig/WithWatchPipeline) ignores delete events because
delete change streams omit fullDocument; update the example to either (A) match
deletes by including conditions on fullDocumentBeforeChange (MongoDB 6.0+ with
pre-images) and/or documentKey (to capture deletes) or (B) add a clear comment
next to the pipeline and WithWatchPipeline call explaining that simple
fullDocument-based filters drop delete events and that consumers must use
fullDocumentBeforeChange or documentKey/pre-images to handle evictions;
reference the pipeline variable, mongo.Pipeline, InitializeWithConfig, and
WithWatchPipeline when making the change.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: caca7689-6c5f-4342-ba40-36c503a11cfb
📒 Files selected for processing (1)
table/cached_generic.go
| // Apply configuration options | ||
| config := &CachedTableConfig{ | ||
| ReadThrough: false, // Default to eager loading | ||
| ReadThrough: false, // Default to eager loading | ||
| Filter: nil, // Default to all entries | ||
| WatchPipeline: nil, // Default to all change events | ||
| } | ||
| for _, opt := range opts { | ||
| opt(config) | ||
| } | ||
| t.readThrough = config.ReadThrough | ||
| t.filter = config.Filter | ||
| t.watchPipeline = config.WatchPipeline |
There was a problem hiding this comment.
Keep filter and watch scope in lockstep.
These options now scope eager load/reconciliation and live updates independently. If a caller sets only one, the table drifts out of scope: WithFilter still ingests unrelated watch events, while WithWatchPipeline can leave part of the eagerly loaded cache stale. Please reject partial configuration or replace the pair with a single higher-level scope option.
Possible guard
for _, opt := range opts {
opt(config)
}
+ if (config.Filter == nil) != (config.WatchPipeline == nil) {
+ return errors.Wrapf(errors.InvalidArgument,
+ "WithFilter and WithWatchPipeline must be configured together to keep cache scope consistent")
+ }
t.readThrough = config.ReadThrough
t.filter = config.Filter
t.watchPipeline = config.WatchPipelineAlso applies to: 194-195, 208-211, 267-270
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@table/cached_generic.go` around lines 162 - 173, After applying opts into
CachedTableConfig, validate that Filter and WatchPipeline are set together: if
exactly one of config.Filter or config.WatchPipeline is non-nil (XOR), reject
the partial configuration (return an error from the construction function or
panic consistently with surrounding API) and surface a clear message referencing
WithFilter/WithWatchPipeline pairing; update the code path that applies opts
(the loop over opts and the assignment to t.readThrough, t.filter,
t.watchPipeline) to perform this check before assigning to
t.filter/t.watchPipeline, and update callers/tests accordingly (or alternatively
collapse them into a single scope option if you prefer that design).
There was a problem hiding this comment.
Validated this finding — partially valid but I disagree with enforcing the pairing as a hard error.
The PR description explains the design decision: filter and pipeline are intentionally separate options because they serve different purposes (initial load vs change stream) and may use different query shapes. There are legitimate use cases for setting only one:
WithFilteralone: Scoped initial load, but accept all change events. Thecallbackhandles out-of-scope events gracefully — if a change event arrives for a document outside the filter scope,DBFindfetches it and it gets added to cache (benign for correctness, just slightly over-inclusive). The reconciler will re-enumerate with the filter on its next sweep.WithWatchPipelinealone: Load all entries upfront but only react to specific change events. Useful when the initial load is cheap but you want to reduce change stream noise.
The common downstream use case (type-scoped isolation in GenericIntegrationTable) will always set both — and the InitializeWithConfig example in the docs already shows them used together.
What I'll do: Add a doc comment on InitializeWithConfig noting that for scoped isolation, both options should typically be set together to avoid cache drift. This makes the guidance explicit without rejecting valid partial configurations at runtime.
There was a problem hiding this comment.
@dev-arya23, your reasoning is sound. The two axes of filtering genuinely can diverge in query shape (a BSON filter document vs. a change-stream $match pipeline), and the two partial-configuration cases you describe are legitimate:
WithFilteronly → over-inclusive cache at most; reconciler sweep brings it back in scope on the next pass.WithWatchPipelineonly → full eager load with reduced change-stream noise; semantically clean.
Adding a documentation note is the right call. Here's a concrete suggestion for the InitializeWithConfig godoc so it's visible at the call site:
// InitializeWithConfig sets up the Table with the provided db.StoreCollection and configuration options.
// ...
//
// # Scoped isolation
//
// For strict scoped isolation (e.g., multiple CachedTable instances sharing one
// collection), set both WithFilter and WithWatchPipeline together so the eager
// load, reconciler enumeration, and change-stream events all cover the same
// document subset. Setting only one option is valid but may cause the cache to
// be slightly over- or under-inclusive relative to the live change stream until
// the reconciler next sweeps.
//
// Example usage:
// ...This keeps the API flexible while making the "use both for isolation" guidance discoverable. Looks good to proceed.
📝 Learnings were identified and not saved (knowledge base disabled). Enable
Prabhjot-Sethi
left a comment
Prabhjot-Sethi
left a comment
There was a problem hiding this comment.
can you check comments provided by coderabbit
- Add delete event warning and fullDocumentBeforeChange example to WithWatchPipeline documentation - Add scope pairing guidance to InitializeWithConfig documentation - Add filter preflight validation in InitializeWithConfig to catch invalid filters at init time with proper error return instead of deferring to panic paths - Convert eager-load FindMany panic to error return since InitializeWithConfig already returns error Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
|
@Prabhjot-Sethi I've reviewed all three CodeRabbit findings and pushed fixes in Finding 1: Watch pipeline drops delete events (line 99) ✅ FixedValid for documentation. The Fix: Updated
Finding 2: Filter and WatchPipeline should be paired (line 173)
|
There was a problem hiding this comment.
♻️ Duplicate comments (1)
table/cached_generic.go (1)
219-227:⚠️ Potential issue | 🟠 MajorReplace
FindMany(SetLimit(0))withCount()for lightweight filter validation.The
FindManycall withSetLimit(0)retrieves and decodes all matching documents from the collection, contradicting the code's claim of "lightweight" validation. On large collections, this will materially impact initialization time and memory usage. UseCount()instead, which validates the filter without fetching document data.Suggested fix
- var preflight []keyOnly[K] - preflightOpts := options.Find().SetLimit(0) - if err := col.FindMany(context.Background(), t.filter, &preflight, preflightOpts); err != nil { + if _, err := col.Count(context.Background(), t.filter); err != nil { return errors.Wrapf(errors.InvalidArgument, "WithFilter: filter validation failed: %s", err) }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@table/cached_generic.go` around lines 219 - 227, The preflight filter validation currently uses col.FindMany with options.Find().SetLimit(0) (see t.filter, preflight, preflightOpts, col.FindMany) which still decodes matching documents; replace this with a lightweight count operation (e.g., col.Count or CountDocuments) that takes the same context and t.filter, check only the error returned, and wrap it with the existing errors.Wrapf message ("WithFilter: filter validation failed: %s"); remove preflight/preflightOpts usage since no result slice is needed. Ensure the new call uses context.Background() and preserves the same error handling path.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@table/cached_generic.go`:
- Around line 219-227: The preflight filter validation currently uses
col.FindMany with options.Find().SetLimit(0) (see t.filter, preflight,
preflightOpts, col.FindMany) which still decodes matching documents; replace
this with a lightweight count operation (e.g., col.Count or CountDocuments) that
takes the same context and t.filter, check only the error returned, and wrap it
with the existing errors.Wrapf message ("WithFilter: filter validation failed:
%s"); remove preflight/preflightOpts usage since no result slice is needed.
Ensure the new call uses context.Background() and preserves the same error
handling path.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: f7ba8fbf-07d2-465a-9440-657ca07d07b3
📒 Files selected for processing (1)
table/cached_generic.go
Addressing CodeRabbit re-review finding:
|
…t validation Count() validates the filter server-side without fetching or decoding any documents, making it a truly lightweight preflight check. SetLimit(0) in MongoDB means "no limit", so the previous FindMany call would actually decode all matching documents — defeating the purpose of a lightweight validation.
|
Pushed fix in All CodeRabbit findings addressed:
@Prabhjot-Sethi — all CodeRabbit findings have been addressed across 3 commits. Ready for your review when you get a chance. |
There was a problem hiding this comment.
🧹 Nitpick comments (1)
table/cached_generic.go (1)
219-228: Good preflight validation — consider preserving error chain.The Count-based preflight is the right approach for lightweight server-side validation. However, the error at line 226 embeds
erras a string via%s, which loses the original error for chain inspection (e.g.,errors.Is/errors.As).If error chain preservation is needed elsewhere in the codebase, consider wrapping the error properly:
♻️ Optional: preserve error chain
if t.filter != nil { if _, err := col.Count(context.Background(), t.filter); err != nil { - return errors.Wrapf(errors.InvalidArgument, "WithFilter: filter validation failed: %s", err) + return errors.Wrapf(err, "WithFilter: filter validation failed") } }Note: This depends on how your custom
errors.Wrapfhandles wrapping — if the first argument is the sentinel error code, the current pattern may be intentional.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@table/cached_generic.go` around lines 219 - 228, The preflight Count error currently embeds err as a string which breaks the error chain; update the WithFilter preflight so the returned error wraps the original err (preserving chain) instead of using %s — e.g., change the errors.Wrapf call around col.Count(context.Background(), t.filter) to use an error-wrapping directive (such as %w) or the equivalent overload of errors.Wrapf/Wrap that accepts the cause so callers can use errors.Is/errors.As to inspect the original error; keep the same message context ("WithFilter: filter validation failed") and reference t.filter, col.Count and errors.Wrapf when making the change.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@table/cached_generic.go`:
- Around line 219-228: The preflight Count error currently embeds err as a
string which breaks the error chain; update the WithFilter preflight so the
returned error wraps the original err (preserving chain) instead of using %s —
e.g., change the errors.Wrapf call around col.Count(context.Background(),
t.filter) to use an error-wrapping directive (such as %w) or the equivalent
overload of errors.Wrapf/Wrap that accepts the cause so callers can use
errors.Is/errors.As to inspect the original error; keep the same message context
("WithFilter: filter validation failed") and reference t.filter, col.Count and
errors.Wrapf when making the change.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 8fe21986-6256-4804-82c0-7721b193b2d3
📒 Files selected for processing (1)
table/cached_generic.go
2 participants
Summary
CachedTable.InitializeWithConfignow accepts two new functional options that allow consumers to scope the cache and change stream to a subset of documents in the collection:WithFilter(filter any)— scopes the initial eager load (FindMany) andReconcilerGetAllKeysto only return entries matching the filter. The filter is passed directly toStoreCollection.FindManyWithWatchPipeline(pipeline any)— scopes the change stream (Watch) to only deliver events matching the pipeline. The pipeline is passed directly toStoreCollection.Watch(expects amongo.Pipeline)Both options are fully backward-compatible — existing callers that pass no options continue to get the current behavior (
nilfilter,nilpipeline = all entries, all events).Motivation
This is the upstream enabler for agentic-core/core#35 (type-scoped isolation for integration tables). The
GenericIntegrationTableinagentic-core/coreembedsCachedTableand needs to scope its cache loading and change stream to a specific integration type (e.g.,"slack","github","email"). Without these options,CachedTablehardcodesnilfor both theWatchpipeline andFindManyfilter, causing every adapter to load all integrations into its cache and receive all change events — which was the root cause of the cross-adapter data corruption bug fixed in email-adapter#10.What changed
CachedTableConfig— two new fields:Filter any— optional filter forFindMany(eager load + reconciler)WatchPipeline any— optional pipeline forWatch(change stream)New option functions:
WithFilter(filter any) CachedTableOptionWithWatchPipeline(pipeline any) CachedTableOptionCachedTablestruct — two new unexported fields:filter any— stored from config, used in eager load andReconcilerGetAllKeyswatchPipeline any— stored from config, passed toWatchInitializeWithConfig— usest.watchPipelineinstead ofnilforWatch, andt.filterinstead ofnilforFindManyReconcilerGetAllKeys— usest.filterinstead ofnilforFindManyHow downstream consumers use this
Design decisions
CachedTableOptionfunctional options pattern already established byWithReadThrough(). No new constructor functions needed, no signature changesTable(non-cached) — the non-cachedTabletype doesn't have eager loading or a reconciler cache, so these options don't apply to itTest plan
go build ./table/...passesgo vet ./table/...passescached_generic_test.gotests pass unchanged (require MongoDB connection)🤖 Generated with Claude Code
Summary by CodeRabbit