Event replay and cleanup
This commit introduces the ability to reconfigure modules/event subscribers by replaying previously stored events at startup. This way modules/subscribers can always come up with the latest activity and data that was present at close at the last run of the program. This process also includes the disposal of unneeded events to minimize disk use over time. The changes include the following: 1. Event interface is extended with the Disposable() function unimplemented stubs for existing interface implementations 2. state.Init() is split into Init() and Start() Init() now initialized the file used by eventStream, and puts in place a 0 size memory cache as a temporary measure on top of the file. Start() reads Pop()s documents one by one from the temp nil-cache eventStream, it attempts to dispose of each event and if the event is not disposable it is pushed onto a second temporary no-memory-cache buffer which is then reverse ordered. The underlying file is truncated and reopened. Finally, the real eventStream is allocated, with in memory cache. All events in the second buffer are published (sent to subscribers as well as added to the new eventStream). 3. Updates to main() to support Init() vs Start() Signed-off-by: Ava Affine <ava@sunnypup.io>
This commit is contained in:
parent
2560410820
commit
2515d396a0
3 changed files with 102 additions and 2 deletions
|
|
@ -40,12 +40,18 @@ const (
|
|||
BadEventObjUnmarshal = "failed to unmarshal event map"
|
||||
BadEventUnmarshal = "failed to unmarshal event"
|
||||
BadEventMissingTypeKey = "event map missing type key"
|
||||
StartBeforeInitError = "state machine not initialized"
|
||||
ReopenStoreFileError = "failed to reopen store file"
|
||||
)
|
||||
|
||||
const (
|
||||
EventTypeMapKey = "type"
|
||||
)
|
||||
|
||||
const (
|
||||
TmpDocBufBackingFile = "bingobot_events_processing"
|
||||
)
|
||||
|
||||
var eventMutex sync.RWMutex
|
||||
var eventSubscriptionCache = [NumEventTypes][]chan Event{}
|
||||
var eventStorageFileName string
|
||||
|
|
@ -60,7 +66,7 @@ func Init(eventMemCacheSize int, eventStoreFileName string) error {
|
|||
}
|
||||
|
||||
if eventStream, err = docbuf.NewDocumentBuffer(
|
||||
eventMemCacheSize,
|
||||
0, // temporary nil-cache buffer to replay events from
|
||||
file,
|
||||
); err != nil {
|
||||
return errors.Join(errors.New(BadEventStreamInit), err)
|
||||
|
|
@ -71,6 +77,79 @@ func Init(eventMemCacheSize int, eventStoreFileName string) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// replay events and begin state machine
|
||||
func Start() error {
|
||||
if eventStream == nil {
|
||||
return errors.New(StartBeforeInitError)
|
||||
}
|
||||
|
||||
tmpBackFile, err := os.CreateTemp("", TmpDocBufBackingFile)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
tmpBuf, err := docbuf.NewDocumentBuffer(0, tmpBackFile)
|
||||
if err != nil {
|
||||
tmpBackFile.Close()
|
||||
return err
|
||||
}
|
||||
|
||||
// filter out disposable events into a reverse ordered on disk stack
|
||||
var doc string
|
||||
for err != nil && err.Error() != docbuf.PopFromEmptyBufferError {
|
||||
doc, err = eventStream.Pop()
|
||||
if err != nil && err.Error() != docbuf.PopFromEmptyBufferError {
|
||||
return err
|
||||
}
|
||||
|
||||
ev, err := EventFromString(doc)
|
||||
if err != nil {
|
||||
logging.Warn("Discarding the following event for not unmarshaling: %s", doc)
|
||||
continue
|
||||
}
|
||||
|
||||
if !ev.Disposable() {
|
||||
tmpBuf.Push(doc)
|
||||
}
|
||||
}
|
||||
|
||||
if err := os.Truncate(eventStorageFileName, 0); err != nil {
|
||||
return errors.Join(errors.New(ReopenStoreFileError), err)
|
||||
}
|
||||
|
||||
file, err := os.OpenFile(eventStorageFileName, os.O_CREATE|os.O_RDWR, 0644)
|
||||
if err != nil {
|
||||
return errors.Join(errors.New(BadEventStoreFilename), err)
|
||||
}
|
||||
|
||||
eventStream, err := docbuf.NewDocumentBuffer(maxEventsInMemory, file)
|
||||
if err != nil {
|
||||
// return error here will panic without truncating or writing to the file
|
||||
return err
|
||||
}
|
||||
|
||||
// unravel tmp stack into properly ordered properly allocated buffer
|
||||
for err != nil && err.Error() != docbuf.PopFromEmptyBufferError {
|
||||
doc, err := eventStream.Pop()
|
||||
if err != nil && err.Error() != docbuf.PopFromEmptyBufferError {
|
||||
logging.Warn("Could not handle the following error: %s", err.Error())
|
||||
continue
|
||||
}
|
||||
|
||||
ev, err := EventFromString(doc)
|
||||
if err != nil {
|
||||
logging.Warn("Could not handle the following error: %s", err.Error())
|
||||
continue
|
||||
}
|
||||
|
||||
if err := PublishEvent(ev); err != nil {
|
||||
logging.Warn("Could not handle the following error: %s", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// return no error. we are handling SIGINT
|
||||
func Teardown() {
|
||||
// riskily ignore all locking....
|
||||
|
|
@ -232,6 +311,9 @@ type Event interface {
|
|||
|
||||
// validates state of internal metadata per EventType
|
||||
Validate() error
|
||||
|
||||
// returns true if event can be discarded
|
||||
Disposable() bool
|
||||
}
|
||||
|
||||
func EventToString(e Event) (string, error) {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue