add persisted watcher library #4307
Conversation
|
Important Review skippedAuto reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the You can disable this status message by setting the TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (invoked as PR comments)
Additionally, you can add CodeRabbit Configuration File (
|
wdbaruni
added
the
build/go
pkg/lib/watcher/watcher.go
Outdated
| return | ||
| } | ||
|
|
||
| w.ctx, w.cancel = context.WithCancel(context.Background()) |
There was a problem hiding this comment.
Please find an alternative to the embedded context (provide it as parameter instead). It's a pattern we need to avoid. https://go.dev/blog/context-and-structs
Contexts should not be stored inside a struct type, but instead passed to each function that needs it.
When designing an API with context, remember the advice: pass context.Context in as an argument; don’t store it in structs.
There was a problem hiding this comment.
Thanks for pointing this out. I understand the concern about storing the provided context and the bad practice mentioned in the link. Though keep in mind I am not storing a provided context as I am creating a fresh one inside Start. My goal is to have better control over how and when background tasks are stopped, which context cancellation doesn't provide.
In this case, the Start() method doesn't accept a context. Instead, I create a new context inside Start to manage the lifecycle of internal calls and to exit any pending GetEvents or HandleEvent calls. Initially, I used a stop channel and a fresh background context in each loop, with the stop channel canceling the context. But this was more complicated than necessary.
I realized I only needed to store the cancel method, not the context itself. Fixed in e6fd1c9
pkg/lib/watcher/boltdb/boltdb.go
Outdated
| err = errors.Join( | ||
| validate.NotNil(db, "boltDB instance cannot be nil"), | ||
| options.validate(), | ||
| ) | ||
| if err != nil { | ||
| return nil, err | ||
| } |
There was a problem hiding this comment.
This check can be performed before the cache and event store are created.
There was a problem hiding this comment.
Correct. Fixed in f74da4d
| case <-s.notifyCh: | ||
| // New events might be available, loop and check again | ||
| // Drain any additional notifications | ||
| for len(s.notifyCh) > 0 { | ||
| <-s.notifyCh | ||
| } |
There was a problem hiding this comment.
What is the purpose of this channel?
There was a problem hiding this comment.
Added a comment in f74da4d
notifyCh is a channel for notifying watchers of new events.
GetEvents will block on this channel when no events are immediately available,
or will return empty events after a long-polling timeout.
Watcher Library
Overview
The Watcher Library is an internal component of the Bacalhau project that provides a robust event watching and processing system. It's designed to efficiently store, retrieve, and process events. The library ensures events are stored in a durable, ordered manner, allowing for consistent and reliable event processing. It supports features like checkpointing, filtering, and long-polling, while maintaining the ability to replay events from any point in the event history.
Key Features
Key Components
Core Concepts
Event
An
Eventrepresents a single occurrence in the system. It has the following properties:SeqNum: A unique, sequential identifier for the event.Operation: The type of operation (Create, Update, Delete).ObjectType: The type of object the event relates to.Object: The actual data associated with the event.Timestamp: When the event occurred.EventStore
The
EventStoreis responsible for persisting events and providing methods to retrieve them. It uses BoltDB as the underlying storage engine and supports features like caching, checkpointing, and garbage collection.Registry
The
Registrymanages multiple watchers. It's the main entry point for components that want to subscribe to events.Watcher
A
Watcherrepresents a single subscriber to events. It processes events sequentially and can be configured with filters and checkpoints.EventIterator
An
EventIteratordefines the starting position for reading events. There are four types of iterators:Usage
Here's how you typically use the Watcher library within Bacalhau:
Configuration
Watch Configuration
When creating a watcher, you can configure it with various options:
WithInitialEventIterator(iterator EventIterator): Sets the starting position for watching if no checkpoint is found.WithFilter(filter EventFilter): Sets the event filter for watching.WithBufferSize(size int): Sets the size of the event buffer.WithBatchSize(size int): Sets the number of events to fetch in each batch.WithInitialBackoff(backoff time.Duration): Sets the initial backoff duration for retries.WithMaxBackoff(backoff time.Duration): Sets the maximum backoff duration for retries.WithMaxRetries(maxRetries int): Sets the maximum number of retries for event handling.WithRetryStrategy(strategy RetryStrategy): Sets the retry strategy for event handling.Example:
EventStore Configuration (BoltDB)
The BoltDB EventStore can be configured with various options:
WithEventsBucket(name string): Sets the name of the bucket used to store events.WithCheckpointBucket(name string): Sets the name of the bucket used to store checkpoints.WithEventSerializer(serializer watcher.Serializer): Sets the serializer used for events.WithCacheSize(size int): Sets the size of the LRU cache used to store events.WithLongPollingTimeout(timeout time.Duration): Sets the timeout duration for long-polling requests.WithGCAgeThreshold(threshold time.Duration): Sets the age threshold for event pruning.WithGCCadence(cadence time.Duration): Sets the interval at which garbage collection runs.WithGCMaxRecordsPerRun(max int): Sets the maximum number of records to process in a single GC run.WithGCMaxDuration(duration time.Duration): Sets the maximum duration for a single GC run.Example:
Best Practices
EventHandlerto ensure robust event processing.Troubleshooting
Future Improvements