Events/cfg enable publisher (#8916)

* only enable publisher based on config

* add default prune tick

* back out state abandon changes on fsm close

nomad/fsm.go

@@ -162,7 +162,6 @@ func NewFSM(config *FSMConfig) (*nomadFSM, error) {
 
 // Close is used to cleanup resources associated with the FSM
 func (n *nomadFSM) Close() error {
-	n.state.Abandon()
 	return nil
 }
 

nomad/state/state_store.go

@@ -45,6 +45,8 @@ type StateStoreConfig struct {
 
 	// Region is the region of the server embedding the state store.
 	Region string
+
+	EnablePublisher bool
 }
 
 // The StateStore is responsible for maintaining all the Nomad
@@ -86,11 +88,16 @@ func NewStateStore(config *StateStoreConfig) (*StateStore, error) {
 		abandonCh:          make(chan struct{}),
 		stopEventPublisher: cancel,
 	}
-	publisher := stream.NewEventPublisher(ctx, stream.EventPublisherCfg{
-		EventBufferTTL:  1 * time.Hour,
-		EventBufferSize: 250,
-	})
-	s.db = NewChangeTrackerDB(db, publisher, processDBChanges)
+
+	if config.EnablePublisher {
+		publisher := stream.NewEventPublisher(ctx, stream.EventPublisherCfg{
+			EventBufferTTL:  1 * time.Hour,
+			EventBufferSize: 250,
+		})
+		s.db = NewChangeTrackerDB(db, publisher, processDBChanges)
+	} else {
+		s.db = NewChangeTrackerDB(db, &noOpPublisher{}, processDBChanges)
+	}
 
 	// Initialize the state store with required enterprise objects
 	if err := s.enterpriseInit(); err != nil {

nomad/stream/event_publisher.go

@@ -55,13 +55,15 @@ func NewEventPublisher(ctx context.Context, cfg EventPublisherCfg) *EventPublish
 	if cfg.EventBufferTTL == 0 {
 		cfg.EventBufferTTL = 1 * time.Hour
 	}
+
 	buffer := newEventBuffer(cfg.EventBufferSize, cfg.EventBufferTTL)
 	e := &EventPublisher{
 		eventBuf:  buffer,
 		publishCh: make(chan changeEvents),
 		subscriptions: &subscriptions{
 			byToken: make(map[string]map[*SubscribeRequest]*Subscription),
 		},
+		pruneTick: 5 * time.Second,
 	}
 
 	go e.handleUpdates(ctx)