Merge pull request #859 from openziti/add.async.events

adds eventual eventer to controller

controller/env/appenv.go

@@ -207,6 +207,10 @@ type AppMiddleware func(*AppEnv, http.Handler) http.Handler
 type authorizer struct {
 }
 
+const (
+	EventualEventsGauge = "eventual.events"
+)
+
 func (a authorizer) Authorize(request *http.Request, principal interface{}) error {
 	//principal is an API Session
 	_, ok := principal.(*model.ApiSession)
@@ -378,6 +382,33 @@ func (ae *AppEnv) InitPersistence() error {
 		return err
 	}
 
+	ae.BoltStores.EventualEventer.AddListener(persistence.EventualEventAddedName, func(i ...interface{}) {
+		if len(i) == 0 {
+			pfxlog.Logger().Errorf("could not update metrics for %s gauge on add, event argument length was 0", EventualEventsGauge)
+			return
+		}
+
+		if event, ok := i[0].(*persistence.EventualEventAdded); ok {
+			gauge := ae.GetHostController().GetNetwork().GetMetricsRegistry().Gauge(EventualEventsGauge)
+			gauge.Update(event.Total)
+		} else {
+			pfxlog.Logger().Errorf("could not update metrics for %s gauge on add, event argument was %T expected *EventualEventAdded", EventualEventsGauge, i[0])
+		}
+	})
+	ae.BoltStores.EventualEventer.AddListener(persistence.EventualEventRemovedName, func(i ...interface{}) {
+		if len(i) == 0 {
+			pfxlog.Logger().Errorf("could not update metrics for %s gauge on remove, event argument length was 0", EventualEventsGauge)
+			return
+		}
+
+		if event, ok := i[0].(*persistence.EventualEventRemoved); ok {
+			gauge := ae.GetHostController().GetNetwork().GetMetricsRegistry().Gauge(EventualEventsGauge)
+			gauge.Update(event.Total)
+		} else {
+			pfxlog.Logger().Errorf("could not update metrics for %s gauge on remove, event argument was %T expected *EventualEventRemoved", EventualEventsGauge, i[0])
+		}
+	})
+
 	ae.Handlers = model.InitHandlers(ae)
 	events.Init(ae.GetDbProvider(), ae.BoltStores, ae.GetHostController().GetCloseNotifyChannel())
 

controller/internal/policy/session_enforcer_test.go

@@ -31,9 +31,8 @@ func Test_SessionEnforcer(t *testing.T) {
 	ctx := &enforcerTestContext{
 		TestContext: model.NewTestContext(t),
 	}
-
-	defer ctx.Cleanup()
 	ctx.Init()
+	defer ctx.Cleanup()
 
 	ctx.testSessionsCleanup()
 }
@@ -75,6 +74,16 @@ func (ctx *enforcerTestContext) testSessionsCleanup() {
 	}
 
 	ctx.NoError(enforcer.Run())
+
+	done, err := ctx.GetStores().EventualEventer.Trigger()
+	ctx.NoError(err)
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		ctx.Fail("did not receive done notification from eventual eventer")
+	}
+
 	ctx.ValidateDeleted(apiSession.Id)
 	ctx.ValidateDeleted(session.Id)
 	ctx.ValidateDeleted(session2.Id)

controller/model/testing.go

@@ -120,10 +120,6 @@ func (ctx *TestContext) GetFingerprintGenerator() cert.FingerprintGenerator {
 }
 
 func NewTestContext(t *testing.T) *TestContext {
-	return NewTestContextWithDb(t, "")
-}
-
-func NewTestContextWithDb(t *testing.T, dbPath string) *TestContext {
 	context := &TestContext{
 		TestContext:     persistence.NewTestContext(t),
 		metricsRegistry: metrics.NewRegistry("test", nil),
@@ -131,9 +127,11 @@ func NewTestContextWithDb(t *testing.T, dbPath string) *TestContext {
 			closeNotify: make(chan struct{}),
 		},
 	}
-	context.InitWithDbFile(dbPath)
 	return context
 }
+func (ctx *TestContext) Init() {
+	ctx.InitWithDbFile("")
+}
 
 func (ctx *TestContext) InitWithDbFile(dbPath string) {
 	ctx.TestContext.InitWithDbFile(dbPath)

controller/persistence/api_session_store.go

@@ -17,7 +17,9 @@
 package persistence
 
 import (
+	"fmt"
 	"github.com/kataras/go-events"
+	"github.com/michaelquigley/pfxlog"
 	"github.com/openziti/edge/eid"
 	"github.com/openziti/foundation/storage/ast"
 	"github.com/openziti/foundation/storage/boltz"
@@ -34,7 +36,8 @@ const (
 	FieldApiSessionMfaRequired    = "mfaRequired"
 	FieldApiSessionLastActivityAt = "lastActivityAt"
 
-	EventFullyAuthenticated events.EventName = "FULLY_AUTHENTICATED"
+	EventFullyAuthenticated       events.EventName = "FULLY_AUTHENTICATED"
+	EventualEventApiSessionDelete                  = "ApiSessionDelete"
 )
 
 type ApiSession struct {
@@ -98,6 +101,8 @@ func newApiSessionStore(stores *stores) *apiSessionStoreImpl {
 	store := &apiSessionStoreImpl{
 		baseStore: newBaseStore(stores, EntityTypeApiSessions),
 	}
+
+	stores.EventualEventer.AddEventualListener(EventualEventApiSessionDelete, store.onEventualDelete)
 	store.InitImpl(store)
 	return store
 }
@@ -109,6 +114,40 @@ type apiSessionStoreImpl struct {
 	symbolIdentity boltz.EntitySymbol
 }
 
+func (store *apiSessionStoreImpl) onEventualDelete(name string, data []byte) {
+	var ids []string
+	err := store.stores.DbProvider.GetDb().View(func(tx *bbolt.Tx) error {
+		query := fmt.Sprintf(`%s = "%s"`, FieldSessionApiSession, string(data))
+		var err error
+		ids, _, err = store.stores.session.QueryIds(tx, query)
+		return err
+	})
+
+	if err != nil {
+		pfxlog.Logger().WithError(err).WithFields(map[string]interface{}{
+			"eventName":    name,
+			"apiSessionId": string(data),
+		}).Error("error querying for session associated to an api session during onEventualDelete")
+	}
+
+	for _, id := range ids {
+		_ = store.stores.DbProvider.GetDb().Update(func(tx *bbolt.Tx) error {
+			ctx := boltz.NewMutateContext(tx)
+			err := store.stores.session.DeleteById(ctx, id)
+
+			if err != nil {
+				pfxlog.Logger().WithError(err).WithFields(map[string]interface{}{
+					"eventName":    name,
+					"apiSessionId": string(data),
+					"sessionId":    id,
+				}).Error("error deleting for session associated to an api session during onEventualDelete")
+			}
+
+			return nil
+		})
+	}
+}
+
 func (store *apiSessionStoreImpl) Create(ctx boltz.MutateContext, entity boltz.Entity) error {
 	err := store.baseStore.Create(ctx, entity)
 
@@ -136,6 +175,20 @@ func (store *apiSessionStoreImpl) Update(ctx boltz.MutateContext, entity boltz.E
 	return err
 }
 
+func (store *apiSessionStoreImpl) DeleteById(ctx boltz.MutateContext, id string) error {
+	err := store.baseStore.DeleteById(ctx, id)
+
+	if err == nil {
+		if bboltEventualEventer, ok := store.baseStore.stores.EventualEventer.(*EventualEventerBbolt); ok {
+			bboltEventualEventer.AddEventualEventWithCtx(ctx, EventualEventApiSessionDelete, []byte(id))
+		} else {
+			store.baseStore.stores.EventualEventer.AddEventualEvent(EventualEventApiSessionDelete, []byte(id))
+		}
+	}
+
+	return err
+}
+
 func (store *apiSessionStoreImpl) NewStoreEntity() boltz.Entity {
 	return &ApiSession{}
 }

controller/persistence/api_session_store_test.go

@@ -39,7 +39,8 @@ func Test_ApiSessionStore(t *testing.T) {
 	t.Run("test delete api sessions", ctx.testDeleteApiSessions)
 }
 
-func (ctx *TestContext) testCreateInvalidApiSessions(_ *testing.T) {
+func (ctx *TestContext) testCreateInvalidApiSessions(t *testing.T) {
+	ctx.BaseTestContext.NextTest(t)
 	defer ctx.CleanupAll()
 
 	apiSession := NewApiSession(eid.New())
@@ -64,7 +65,8 @@ func (ctx *TestContext) testCreateInvalidApiSessions(_ *testing.T) {
 	ctx.EqualError(err, fmt.Sprintf("an entity of type apiSession already exists with id %v", apiSession.GetId()))
 }
 
-func (ctx *TestContext) testCreateApiSessions(_ *testing.T) {
+func (ctx *TestContext) testCreateApiSessions(t *testing.T) {
+	ctx.BaseTestContext.NextTest(t)
 	ctx.CleanupAll()
 
 	identity := ctx.RequireNewIdentity("Jojo", false)
@@ -80,9 +82,20 @@ func (ctx *TestContext) testCreateApiSessions(_ *testing.T) {
 
 	ctx.ValidateBaseline(apiSession2)
 
-	ctx.RequireDelete(apiSession)
-
+	err := ctx.Delete(apiSession)
+	ctx.NoError(err)
 	ctx.RequireDelete(identity)
+
+	done, err := ctx.GetStores().EventualEventer.Trigger()
+	ctx.NoError(err)
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		ctx.Fail("did not receive done notification from eventual eventer")
+
+	}
+
 	ctx.ValidateDeleted(apiSession.Id)
 	ctx.ValidateDeleted(apiSession2.Id)
 }
@@ -130,7 +143,8 @@ func (ctx *TestContext) createApiSessionTestEntities() *apiSessionTestEntities {
 	}
 }
 
-func (ctx *TestContext) testLoadQueryApiSessions(_ *testing.T) {
+func (ctx *TestContext) testLoadQueryApiSessions(t *testing.T) {
+	ctx.BaseTestContext.NextTest(t)
 	ctx.CleanupAll()
 
 	entities := ctx.createApiSessionTestEntities()
@@ -159,11 +173,11 @@ func (ctx *TestContext) testLoadQueryApiSessions(_ *testing.T) {
 	ctx.NoError(err)
 }
 
-func (ctx *TestContext) testUpdateApiSessions(_ *testing.T) {
+func (ctx *TestContext) testUpdateApiSessions(t *testing.T) {
+	ctx.BaseTestContext.NextTest(t)
 	ctx.CleanupAll()
 	entities := ctx.createApiSessionTestEntities()
 	earlier := time.Now()
-	time.Sleep(time.Millisecond * 50)
 
 	err := ctx.GetDb().Update(func(tx *bbolt.Tx) error {
 		original, err := ctx.stores.ApiSession.LoadOneById(tx, entities.apiSession1.Id)
@@ -197,10 +211,31 @@ func (ctx *TestContext) testUpdateApiSessions(_ *testing.T) {
 	ctx.NoError(err)
 }
 
-func (ctx *TestContext) testDeleteApiSessions(_ *testing.T) {
+func (ctx *TestContext) testDeleteApiSessions(t *testing.T) {
+	ctx.BaseTestContext.NextTest(t)
 	ctx.CleanupAll()
 	entities := ctx.createApiSessionTestEntities()
-	ctx.RequireDelete(entities.apiSession1)
-	ctx.RequireDelete(entities.apiSession2)
-	ctx.RequireDelete(entities.apiSession3)
+
+	err := ctx.Delete(entities.apiSession1)
+	ctx.NoError(err)
+
+	err = ctx.Delete(entities.apiSession2)
+	ctx.NoError(err)
+
+	err = ctx.Delete(entities.apiSession3)
+	ctx.NoError(err)
+
+	done, err := ctx.GetStores().EventualEventer.Trigger()
+	ctx.NoError(err)
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		ctx.Fail("did not receive done notification from eventual eventer")
+
+	}
+
+	ctx.ValidateDeleted(entities.apiSession1.GetId())
+	ctx.ValidateDeleted(entities.apiSession2.GetId())
+	ctx.ValidateDeleted(entities.apiSession3.GetId())
 }

controller/persistence/base_entity.go

@@ -24,6 +24,7 @@ import (
 const (
 	EntityTypeApiSessions               = "apiSessions"
 	EntityTypeApiSessionCertificates    = "apiSessionCertificates"
+	EntityTypeEventualEvents            = "eventualEvents"
 	EntityTypeCas                       = "cas"
 	EntityTypeConfigs                   = "configs"
 	EntityTypeConfigTypes               = "configTypes"

controller/persistence/eventual_event_store.go

@@ -0,0 +1,99 @@
+/*
+	Copyright NetFoundry, Inc.
+
+	Licensed under the Apache License, Version 2.0 (the "License");
+	you may not use this file except in compliance with the License.
+	You may obtain a copy of the License at
+
+	https://www.apache.org/licenses/LICENSE-2.0
+
+	Unless required by applicable law or agreed to in writing, software
+	distributed under the License is distributed on an "AS IS" BASIS,
+	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+	See the License for the specific language governing permissions and
+	limitations under the License.
+*/
+
+package persistence
+
+import (
+	"github.com/openziti/foundation/storage/ast"
+	"github.com/openziti/foundation/storage/boltz"
+	"go.etcd.io/bbolt"
+)
+
+const (
+	FieldEventualEventType = "type"
+	FieldEventualEventData = "data"
+)
+
+type EventualEvent struct {
+	boltz.BaseExtEntity
+	Type string
+	Data []byte
+}
+
+func (entity *EventualEvent) LoadValues(_ boltz.CrudStore, bucket *boltz.TypedBucket) {
+	entity.LoadBaseValues(bucket)
+	entity.Type = bucket.GetStringOrError(FieldEventualEventType)
+	entity.Data = bucket.Get([]byte(FieldEventualEventData))
+
+}
+
+func (entity *EventualEvent) SetValues(ctx *boltz.PersistContext) {
+	entity.SetBaseValues(ctx)
+	ctx.SetString(FieldEventualEventType, entity.Type)
+	ctx.Bucket.SetError(ctx.Bucket.Put([]byte(FieldEventualEventData), entity.Data))
+}
+
+func (entity *EventualEvent) GetEntityType() string {
+	return EntityTypeEventualEvents
+}
+
+type EventualEventStore interface {
+	Store
+	LoadOneById(tx *bbolt.Tx, id string) (*EventualEvent, error)
+	LoadOneByQuery(tx *bbolt.Tx, query string) (*EventualEvent, error)
+}
+
+func newEventualEventStore(stores *stores) *eventualEventStoreImpl {
+	store := &eventualEventStoreImpl{
+		baseStore: newBaseStore(stores, EntityTypeEventualEvents),
+	}
+	store.InitImpl(store)
+	return store
+}
+
+type eventualEventStoreImpl struct {
+	*baseStore
+	indexName         boltz.ReadIndex
+	symbolEnrollments boltz.EntitySetSymbol
+}
+
+func (store *eventualEventStoreImpl) LoadOneById(tx *bbolt.Tx, id string) (*EventualEvent, error) {
+	entity := &EventualEvent{}
+	if err := store.baseLoadOneById(tx, id, entity); err != nil {
+		return nil, err
+	}
+	return entity, nil
+}
+
+func (store *eventualEventStoreImpl) LoadOneByQuery(tx *bbolt.Tx, query string) (*EventualEvent, error) {
+	entity := &EventualEvent{}
+	if found, err := store.BaseLoadOneByQuery(tx, query, entity); !found || err != nil {
+		return nil, err
+	}
+	return entity, nil
+}
+
+func (store *eventualEventStoreImpl) NewStoreEntity() boltz.Entity {
+	return &EventualEvent{}
+}
+
+func (store *eventualEventStoreImpl) initializeLocal() {
+	store.AddExtEntitySymbols()
+	store.AddSymbol(FieldEventualEventData, ast.NodeTypeOther)
+}
+
+func (store *eventualEventStoreImpl) initializeLinked() {
+}

controller/persistence/eventual_eventer.go

@@ -0,0 +1,696 @@
+/*
+	Copyright NetFoundry, Inc.
+
+	Licensed under the Apache License, Version 2.0 (the "License");
+	you may not use this file except in compliance with the License.
+	You may obtain a copy of the License at
+
+	https://www.apache.org/licenses/LICENSE-2.0
+
+	Unless required by applicable law or agreed to in writing, software
+	distributed under the License is distributed on an "AS IS" BASIS,
+	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+	See the License for the specific language governing permissions and
+	limitations under the License.
+*/
+
+package persistence
+
+import (
+	"fmt"
+	"github.com/kataras/go-events"
+	"github.com/lucsky/cuid"
+	"github.com/michaelquigley/pfxlog"
+	"github.com/openziti/foundation/storage/boltz"
+	"github.com/openziti/foundation/util/concurrenz"
+	cmap "github.com/orcaman/concurrent-map"
+	"github.com/pkg/errors"
+	"go.etcd.io/bbolt"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// An EventualEventer provides a method for storing events in a persistent manner
+// that will be processed at a later date. Processing may include time intensive processing such
+// as bulk deletion of other entities. Event persistence strategy, processing order, and processing
+// synchronization are up to the implementation to decide.
+//
+// EventualEventers are also required to emit a series of events via the events.EventEmitter
+// interface. See EventualEventAdded and subsequent events for more details.
+type EventualEventer interface {
+	// EventEmmiter is used to provide processing event status on processing state, which is useful
+	// for instrumenting an EventualEventer for metric purposes (process runtime, process batch runtime,
+	// event counts, etc.)
+	events.EventEmmiter
+
+	// AddEventualEvent adds an eventual event with a specific name and byte array data payload. Interpretation
+	// of the event's data payload is upto the event emitter and consumer.
+	AddEventualEvent(eventType string, data []byte)
+
+	// AddEventualListener adds a function as call back when an eventual event is processed.
+	AddEventualListener(eventType string, handler EventListenerFunc)
+
+	// Start should be called at the start of the lifetime of the EventualEventer.
+	// A closeNotify channel must be supplied for application shutdown eventing.
+	//
+	// If an EventualEventer has already been started, it will return an error.
+	// Errors may be returned for other reasons causing Start to fail.
+	Start(closeNotify <-chan struct{}) error
+
+	// Stop may be called to manually end of the lifetime of the EventualEventer outside the
+	// closeNotify signaling provided in the Start call. If not started, an error will be returned.
+	// Errors may be returned for other reasons causing Stop to fail.
+	Stop() error
+
+	// Trigger forces an EventualEventer to check for work to be processed. Beyond this method,
+	// it is the implementation's responsibility to provide other mechanisms or logic to determine
+	// when work is performed (timers, events, etc.) which may be setup/torn down during Start/Stop.
+	//
+	// If the EventualEventer is not currently running or can't process work and error will
+	// be returned. If it is running a channel will be returned which will be closed after
+	// the current or next iteration of the event processor has completed.
+	Trigger() (<-chan struct{}, error)
+}
+
+// EventListenerFunc is a function handler that will be triggered asynchronously some point in the future
+type EventListenerFunc func(name string, data []byte)
+
+type EventualEventAdded struct {
+	// Id is a unique id for the event created
+	Id string
+
+	// Total is the total number of eventual events awaiting processing
+	Total int64
+}
+
+type EventualEventRemoved struct {
+	// Id is a unique id for the event deleted
+	Id string
+
+	// Total is the total number of eventual events awaiting processing
+	Total int64
+}
+
+type EventualEventProcessingStart struct {
+	// Id is a unique id for processing run
+	Id string
+
+	// StartTime is the time the processing began
+	StartTime time.Time
+}
+
+type EventualEventProcessingBatchStart struct {
+	// Id is a unique id for the batch
+	Id string
+
+	// Id is the unique processing run this batch is a member of
+	ProcessId string
+
+	// Count is the number of events in the current batch
+	Count int
+
+	// BatchSize is the batch size for the current batch (the maximum value of Count)
+	BatchSize int
+
+	// StartTime the time when the batch started processing
+	StartTime time.Time
+}
+
+type EventualEventProcessingListenerStart struct {
+	// Id is a unique id for the triggering of a listener
+	Id string
+
+	// BatchId is the unique id of the batch being processed
+	BatchId string
+
+	// ProcessId is the unique id of the currently executing process
+	ProcessId string
+
+	// ListenerFunc is the listener that was executed
+	ListenerFunc EventListenerFunc
+
+	// BatchEventIndex is the zero based offset of the currently executing event
+	BatchEventIndex int64
+
+	// TotalEventIndex is the total index across all batches of the currently executing event
+	TotalEventIndex int64
+
+	// EventType is the typeof the event that is triggering the listener
+	EventType string
+
+	// StartTime is the time when the listener was started
+	StartTime time.Time
+}
+
+type EventualEventProcessingListenerDone struct {
+	// Id is a unique id for the triggering of a listener
+	Id string
+
+	// BatchId is the unique id of the batch being processed
+	BatchId string
+
+	// ProcessId is the unique id of the currently executing process
+	ProcessId string
+
+	// ListenerFunc is the listener that was executed
+	ListenerFunc EventListenerFunc
+
+	// BatchEventIndex is the zero based offset of the currently executing event
+	BatchEventIndex int64
+
+	// TotalEventIndex is the total index across all batches of the currently executing event
+	TotalEventIndex int64
+
+	// Error is nil if no error occurred during execution, otherwise an error value
+	Error error
+
+	// EventType is the typeof the event that triggered the listener
+	EventType string
+
+	// StartTime is the time when the listener started execution
+	StartTime time.Time
+
+	// EndTime is the time when the listener ended execution
+	EndTime time.Time
+}
+
+type EventualEventProcessingBatchDone struct {
+	// Id is a unique id for the batch
+	Id string
+
+	// Id is the unique processing run this batch is a member of
+	ProcessId string
+
+	// Count is the number of events in the current batch
+	Count int
+
+	// BatchSize is the batch size for the current batch (the maximum value of Count)
+	BatchSize int
+
+	// StartTime the time the batch was started
+	StartTime time.Time
+
+	// EndTime the time the batch ended
+	EndTime time.Time
+}
+
+type EventualEventProcessingDone struct {
+	// Id is a unique id for processing run
+	Id string
+
+	// TotalBatches is the total number of batches executed during processing
+	TotalBatches int64
+
+	// TotalEvent is the total number of events processed
+	TotalEvents int64
+
+	// TotalListenersExecuted is the total number of listeners executed during processing
+	TotalListenersExecuted int64
+
+	// StartTime is the time when the processing began
+	StartTime time.Time
+
+	// EndTime is the time when the processing ended
+	EndTime time.Time
+}
+
+const (
+	// EventualEventAddedName is emitted when a new event is added via AddEventualEvent().
+	//
+	// Event arguments:
+	//	0 - an EventualEventAdded struct
+	EventualEventAddedName = events.EventName("EventualEventAdded")
+
+	// EventualEventRemovedName is emitted when a previously added eventual event is processed
+	//
+	// Event arguments:
+	//	0 - an EventualEventRemoved struct
+	EventualEventRemovedName = events.EventName("EventualEventRemoved")
+
+	// EventualEventProcessingStartName is emitted as the first action during processing
+	// Event arguments:
+	//	0 - an EventualEventProcessingStart struct
+	EventualEventProcessingStartName = events.EventName("EventualEventProcessingStart")
+
+	// EventualEventProcessingBatchStartName is emitted as the first set of events are processed
+	// after EventualEventProcessingStartName. It is possible for 0+ batches to be processed. Each
+	// patch should contain 1+ events.
+	//
+	// Event arguments:
+	//	0 - an EventualEventProcessingBatchStart struct
+	EventualEventProcessingBatchStartName = events.EventName("EventualEventProcessingBatchStart")
+
+	// EventualEventProcessingListenerStartName is emitted for each function listener invoked
+	// on each event.
+	//
+	// Event arguments:
+	//	0 - an EventualEventProcessingListenerStart struct
+	EventualEventProcessingListenerStartName = events.EventName("EventualEventProcessingListenerStart")
+
+	// EventualEventProcessingListenerDoneName is emitted for each function listener after invocation
+	//
+	// Event arguments:
+	//	0 - an EventualEventProcessingListenerDone struct
+	EventualEventProcessingListenerDoneName = events.EventName("EventualEventProcessingListenerDone")
+
+	// EventualEventProcessingBatchDoneName is emitted after the last event processed in a batch.
+	//
+	// Event arguments:
+	//	0 - an EventualEventProcessingBatchDone struct
+	EventualEventProcessingBatchDoneName = events.EventName("EventualEventProcessingBatchDone")
+
+	// EventualEventProcessingDoneName is emitted as the last action during processing after
+	// all events and batches.
+	//
+	// Event arguments:
+	//	0 - an EventualEventProcessingDone struct
+	EventualEventProcessingDoneName = events.EventName("EventualEventProcessingDone")
+)
+
+// EventualEventerBbolt implements EventualEventer with a bbolt back storage mechanism.
+// Work is performed on a configurable basis via the Interval property in FIFO order.
+//
+// Events are stored in the following format:
+//		id   - CUID   - a monotonic reference id
+//      name - string - an event name, used for log output
+//      data - []byte - a string array of arguments
+type EventualEventerBbolt struct {
+	events.EventEmmiter
+	handlerMap        cmap.ConcurrentMap //eventName -> handler func(name, []byte)
+	Interval          time.Duration
+	closeNotify       <-chan struct{}
+	stopNotify        chan struct{}
+	trigger           chan struct{}
+	outstandingEvents *int64
+
+	waiters sync.Map //id -> chan struct{}
+
+	running    concurrenz.AtomicBoolean
+	batchSize  int
+	dbProvider DbProvider
+	store      EventualEventStore
+}
+
+var _ EventualEventer = &EventualEventerBbolt{}
+
+// NewEventualEventerBbolt creates a new bbolt backed asynchronous eventer that will check for new events at the given interval
+// or when triggered. On each interval/trigger, the number of events processed is determined by batchSize.
+func NewEventualEventerBbolt(dbProvider DbProvider, store EventualEventStore, interval time.Duration, batchSize int) *EventualEventerBbolt {
+	outstanding := int64(0)
+	return &EventualEventerBbolt{
+		EventEmmiter:      events.New(),
+		Interval:          interval,
+		dbProvider:        dbProvider,
+		store:             store,
+		batchSize:         batchSize,
+		trigger:           make(chan struct{}, 1),
+		handlerMap:        cmap.New(),
+		outstandingEvents: &outstanding,
+	}
+}
+
+func (a *EventualEventerBbolt) AddEventualEventWithCtx(ctx boltz.MutateContext, eventType string, data []byte) {
+	newId := cuid.New()
+	total := atomic.AddInt64(a.outstandingEvents, 1)
+	a.Emit(EventualEventAddedName, &EventualEventAdded{
+		Id:    newId,
+		Total: total,
+	})
+
+	event := &EventualEvent{
+		BaseExtEntity: boltz.BaseExtEntity{
+			Id:        newId,
+			CreatedAt: time.Now(),
+			UpdatedAt: time.Now(),
+		},
+		Type: eventType,
+		Data: data,
+	}
+
+	var err error
+	if ctx == nil {
+		err = a.dbProvider.GetDb().Update(func(tx *bbolt.Tx) error {
+			ctx := boltz.NewMutateContext(tx)
+			return a.store.Create(ctx, event)
+		})
+	} else {
+		err = a.store.Create(ctx, event)
+	}
+
+	if err != nil {
+		total := atomic.AddInt64(a.outstandingEvents, -1)
+		a.Emit(EventualEventRemovedName, &EventualEventRemoved{
+			Id:    newId,
+			Total: total,
+		})
+		pfxlog.Logger().WithError(err).Error("error adding event for EventualEventerBbolt")
+		return
+	}
+}
+
+func (a *EventualEventerBbolt) AddEventualEvent(eventType string, data []byte) {
+	a.AddEventualEventWithCtx(nil, eventType, data)
+}
+
+func (a *EventualEventerBbolt) AddEventualListener(eventType string, listener EventListenerFunc) {
+	a.handlerMap.Upsert(eventType, nil, func(exist bool, valueInMap interface{}, newValue interface{}) interface{} {
+		var handlers []EventListenerFunc
+		if exist {
+			handlers = valueInMap.([]EventListenerFunc)
+		}
+
+		handlers = append(handlers, listener)
+		return handlers
+	})
+}
+
+func (a *EventualEventerBbolt) Start(closeNotify <-chan struct{}) error {
+	if !a.running.CompareAndSwap(false, true) {
+		return errors.New("already started")
+	}
+	a.stopNotify = make(chan struct{}, 0)
+	a.closeNotify = closeNotify
+	go a.run()
+
+	return nil
+}
+
+// run should be executed once on start and is the main
+// processing loop. Exiting is based on close/stop notify
+// channels provided during creation. Processing is triggered
+// when Trigger() is called or on the configured interval.
+func (a *EventualEventerBbolt) run() {
+	for {
+		select {
+		case <-a.closeNotify:
+			a.running.Set(false)
+			return
+		case <-a.stopNotify:
+			a.running.Set(false)
+			return
+		case <-a.trigger:
+			a.process()
+		case <-time.After(a.Interval):
+			a.trigger <- struct{}{}
+		}
+	}
+}
+
+func (a *EventualEventerBbolt) Stop() error {
+	if a.running.CompareAndSwap(true, false) {
+		close(a.stopNotify)
+		return nil
+	}
+
+	return nil
+}
+
+func (a *EventualEventerBbolt) Trigger() (<-chan struct{}, error) {
+	if a.running.Get() {
+		doneNotify := make(chan struct{})
+		a.waiters.Store(cuid.New(), doneNotify)
+
+		select {
+		case a.trigger <- struct{}{}:
+		default:
+			//channel full, already queued for processing
+		}
+
+		return doneNotify, nil
+	}
+
+	return nil, errors.New("triggering is impossible, eventual eventer not started")
+}
+
+// deleteEventualEvent removes an eventual event by id from the bbolt backend store.
+func (a *EventualEventerBbolt) deleteEventualEvent(id string) error {
+	err := a.dbProvider.GetDb().Update(func(tx *bbolt.Tx) error {
+		ctx := boltz.NewMutateContext(tx)
+		return a.store.DeleteById(ctx, id)
+	})
+
+	if err == nil {
+		total := atomic.AddInt64(a.outstandingEvents, -1)
+		a.Emit(EventualEventRemovedName, &EventualEventRemoved{
+			Id:    id,
+			Total: total,
+		})
+	}
+
+	return err
+}
+
+// notifyWaiters closes all signaling channels returned in calls
+// to Trigger().
+func (a *EventualEventerBbolt) notifyWaiters() {
+	var waiterIds []string
+
+	a.waiters.Range(func(key, value interface{}) bool {
+		id := key.(string)
+		waiterChan := value.(chan struct{})
+		if waiterChan != nil {
+			close(waiterChan)
+		}
+
+		waiterIds = append(waiterIds, id)
+		return true
+	})
+
+	for _, id := range waiterIds {
+		a.waiters.Delete(id)
+	}
+}
+
+// getEventualEvents returns eventual events up from the persistent bbolt store. The number
+// returned is determined by batchSize and the order of the events are based on key sorting
+func (a *EventualEventerBbolt) getEventualEvents() ([]string, []*EventualEvent, error) {
+	var ids []string
+	var eventualEvents []*EventualEvent
+	err := a.dbProvider.GetDb().View(func(tx *bbolt.Tx) error {
+		var err error
+		ids, _, err = a.store.QueryIds(tx, fmt.Sprintf("limit %d", a.batchSize))
+
+		if err != nil {
+			return err
+		}
+
+		for _, id := range ids {
+			event, err := a.store.LoadOneById(tx, id)
+
+			if err != nil {
+				pfxlog.Logger().WithField("id", id).WithError(err).Errorf("error could not load event id %s", id)
+			} else {
+				eventualEvents = append(eventualEvents, event)
+			}
+		}
+		return nil
+	})
+
+	return ids, eventualEvents, err
+}
+
+type runInfo struct {
+	processId              string
+	totalBatches           int64
+	totalEvents            int64
+	totalListenersExecuted int64
+	totalEventIndex        int64
+	batchEventIndex        int64
+	batchId                string
+	numIds                 int
+}
+
+// resolveEventualEvents will wait if there are newly added eventual events
+// that have a transaction that has not cleared the event store yet. This covers
+// situations where tests are expecting the next run to resolve their event. resolveEvents
+// intentionally adds 2500 microseconds (2.5ms) when new events have been added, but
+// none are returned when queried
+func (a *EventualEventerBbolt) resolveEventualEvents() (int, []string, []*EventualEvent) {
+	var numIds int
+	var ids []string
+	var eventualEvents []*EventualEvent
+
+	// if we know we have outstanding events query for them up to 5 times then give up
+	numEvents := atomic.LoadInt64(a.outstandingEvents)
+
+	for attempt := 0; attempt < 5 && numEvents > 0; attempt++ {
+		var err error
+		ids, eventualEvents, err = a.getEventualEvents()
+		numIds = len(ids)
+
+		if err != nil {
+			pfxlog.Logger().WithError(err).Error("error during EventualEventerBbolt processing, could not query events")
+		}
+
+		if numIds > 0 {
+			break
+		}
+
+		time.Sleep(500 * time.Microsecond)
+		numEvents = atomic.LoadInt64(a.outstandingEvents)
+	}
+
+	return numIds, ids, eventualEvents
+}
+
+// process is the main execution look for dealing with batches of eventual events, triggering listeners for
+// the eventual events and emitting normal processing events
+func (a *EventualEventerBbolt) process() {
+
+	info := &runInfo{
+		processId: cuid.New(),
+	}
+
+	startTime := time.Now()
+
+	a.Emit(EventualEventProcessingStartName, &EventualEventProcessingStart{
+		Id:        info.processId,
+		StartTime: startTime,
+	})
+
+	defer func() {
+		a.notifyWaiters()
+
+		a.Emit(EventualEventProcessingDoneName, &EventualEventProcessingDone{
+			Id:                     info.processId,
+			TotalBatches:           info.totalBatches,
+			TotalEvents:            info.totalEvents,
+			TotalListenersExecuted: info.totalListenersExecuted,
+			StartTime:              startTime,
+			EndTime:                time.Now(),
+		})
+	}()
+
+	for {
+		info.batchId = cuid.New()
+		info.numIds = 0
+
+		var ids []string
+		var eventualEvents []*EventualEvent
+
+		info.numIds, ids, eventualEvents = a.resolveEventualEvents()
+
+		if info.numIds == 0 {
+			return
+		}
+
+		if info.numIds != len(eventualEvents) {
+			pfxlog.Logger().
+				WithFields(map[string]interface{}{
+					"queriedCount": info.numIds,
+					"eventCount":   len(eventualEvents),
+				}).
+				Warnf("%d event records were queried and only %d events were loaded", info.numIds, len(eventualEvents))
+		}
+
+		a.processBatch(info, eventualEvents)
+		a.deleteEvents(ids)
+	}
+}
+
+func (a *EventualEventerBbolt) processBatch(info *runInfo, eventualEvents []*EventualEvent) {
+	info.totalBatches++
+	startTime := time.Now()
+	a.Emit(EventualEventProcessingBatchStartName, &EventualEventProcessingBatchStart{
+		Id:        info.batchId,
+		ProcessId: info.processId,
+		Count:     info.numIds,
+		BatchSize: a.batchSize,
+		StartTime: startTime,
+	})
+
+	info.batchEventIndex = 0
+	for _, eventualEvent := range eventualEvents {
+		info.batchEventIndex++
+		info.totalEventIndex++
+
+		if handlerVals, found := a.handlerMap.Get(eventualEvent.Type); found {
+			handlers := handlerVals.([]EventListenerFunc)
+			for _, handler := range handlers {
+				a.executeHandler(info, eventualEvent, handler)
+			}
+		} else {
+			pfxlog.Logger().
+				WithFields(map[string]interface{}{
+					"id":   eventualEvent.Id,
+					"type": eventualEvent.Type,
+				}).
+				Debugf("event id %s with type %s has no handlers", eventualEvent.Id, eventualEvent.Type)
+		}
+	}
+
+	a.Emit(EventualEventProcessingBatchDoneName, &EventualEventProcessingBatchDone{
+		Id:        info.batchId,
+		ProcessId: info.processId,
+		Count:     info.numIds,
+		BatchSize: a.batchSize,
+		StartTime: startTime,
+		EndTime:   time.Now(),
+	})
+}
+
+func (a *EventualEventerBbolt) deleteEvents(ids []string) {
+	for _, id := range ids {
+		if err := a.deleteEventualEvent(id); err != nil {
+			pfxlog.Logger().WithError(err).
+				WithField("id", id).
+				Errorf("could not delete event id %s", id)
+		}
+	}
+}
+
+func (a *EventualEventerBbolt) executeHandler(info *runInfo, eventualEvent *EventualEvent, handler EventListenerFunc) {
+	listenerExecId := cuid.New()
+	info.totalListenersExecuted++
+
+	var startTime time.Time
+	var endTime time.Time
+
+	defer func() {
+		endTime = time.Now()
+
+		var err error
+		if r := recover(); r != nil {
+
+			switch x := r.(type) {
+			case string:
+				err = errors.New(x)
+			case error:
+				err = x
+			default:
+				err = errors.New("unknown panic")
+			}
+
+			pfxlog.Logger().WithError(err).Errorf("panic caught during asynchronous event %s handler %v", eventualEvent.Type, handler)
+		}
+
+		a.Emit(EventualEventProcessingListenerDoneName, &EventualEventProcessingListenerDone{
+			Id:              listenerExecId,
+			BatchId:         info.batchId,
+			ProcessId:       info.processId,
+			ListenerFunc:    handler,
+			BatchEventIndex: info.batchEventIndex,
+			TotalEventIndex: info.totalEventIndex,
+			EventType:       eventualEvent.Type,
+			Error:           err,
+			StartTime:       startTime,
+			EndTime:         endTime,
+		})
+	}()
+
+	startTime = time.Now()
+
+	a.Emit(EventualEventProcessingListenerStartName, &EventualEventProcessingListenerStart{
+		Id:              listenerExecId,
+		BatchId:         info.batchId,
+		ProcessId:       info.processId,
+		ListenerFunc:    handler,
+		BatchEventIndex: info.batchEventIndex,
+		TotalEventIndex: info.totalEventIndex,
+		EventType:       eventualEvent.Type,
+		StartTime:       startTime,
+	})
+
+	handler(eventualEvent.Type, eventualEvent.Data)
+}

controller/persistence/session_store.go

@@ -179,7 +179,7 @@ func (store *sessionStoreImpl) initializeLocal() {
 	store.symbolServicePolicies = store.AddFkSetSymbol(FieldSessionServicePolicies, store.stores.servicePolicy)
 	store.AddSymbol(FieldSessionType, ast.NodeTypeString)
 
-	store.AddFkConstraint(store.symbolApiSession, false, boltz.CascadeDelete)
+	store.AddFkConstraint(store.symbolApiSession, false, boltz.CascadeCreateUpdate)
 	store.AddFkConstraint(store.symbolService, false, boltz.CascadeDelete)
 }
 

controller/persistence/session_store_test.go

@@ -135,7 +135,21 @@ func (ctx *TestContext) testCreateSessions(_ *testing.T) {
 	ctx.RequireReload(session)
 	ctx.RequireReload(session2)
 
-	ctx.RequireDelete(apiSession)
+	err := ctx.Delete(apiSession)
+	ctx.NoError(err)
+
+	done, err := ctx.GetStores().EventualEventer.Trigger()
+	ctx.NoError(err)
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		ctx.Fail("did not receive done notification from eventual eventer")
+
+	}
+
+	ctx.ValidateDeleted(apiSession.GetId())
+
 	ctx.ValidateDeleted(session.Id)
 	ctx.ValidateDeleted(session2.Id)
 

controller/persistence/stores.go

@@ -25,10 +25,12 @@ import (
 	"github.com/pkg/errors"
 	"go.etcd.io/bbolt"
 	"reflect"
+	"time"
 )
 
 type Stores struct {
-	DbProvider DbProvider
+	DbProvider      DbProvider
+	EventualEventer EventualEventer
 
 	// fabric stores
 	Router     db.RouterStore
@@ -37,6 +39,7 @@ type Stores struct {
 
 	ApiSession              ApiSessionStore
 	ApiSessionCertificate   ApiSessionCertificateStore
+	EventualEvent           EventualEventStore
 	Ca                      CaStore
 	Config                  ConfigStore
 	ConfigType              ConfigTypeStore
@@ -116,7 +119,8 @@ func (stores *Stores) CheckIntegrityInTx(tx *bbolt.Tx, fix bool, errorHandler fu
 }
 
 type stores struct {
-	DbProvider DbProvider
+	DbProvider      DbProvider
+	EventualEventer EventualEventer
 
 	// fabric stores
 	Router     db.RouterStore
@@ -124,6 +128,7 @@ type stores struct {
 	Terminator db.TerminatorStore
 
 	apiSession              *apiSessionStoreImpl
+	eventualEvent           *eventualEventStoreImpl
 	ca                      *caStoreImpl
 	config                  *configStoreImpl
 	configType              *configTypeStoreImpl
@@ -153,10 +158,12 @@ func NewBoltStores(dbProvider DbProvider) (*Stores, error) {
 		DbProvider: dbProvider,
 	}
 
+	internalStores.eventualEvent = newEventualEventStore(internalStores)
+	internalStores.EventualEventer = NewEventualEventerBbolt(dbProvider, internalStores.eventualEvent, 2*time.Second, 1000)
+
 	internalStores.Terminator = dbProvider.GetStores().Terminator
 	internalStores.Router = dbProvider.GetStores().Router
 	internalStores.Service = dbProvider.GetStores().Service
-
 	internalStores.apiSession = newApiSessionStore(internalStores)
 	internalStores.apiSessionCertificate = newApiSessionCertificateStore(internalStores)
 	internalStores.authenticator = newAuthenticatorStore(internalStores)
@@ -188,6 +195,7 @@ func NewBoltStores(dbProvider DbProvider) (*Stores, error) {
 
 		ApiSession:              internalStores.apiSession,
 		ApiSessionCertificate:   internalStores.apiSessionCertificate,
+		EventualEvent:           internalStores.eventualEvent,
 		Ca:                      internalStores.ca,
 		Config:                  internalStores.config,
 		ConfigType:              internalStores.configType,
@@ -211,6 +219,8 @@ func NewBoltStores(dbProvider DbProvider) (*Stores, error) {
 		storeMap: make(map[reflect.Type]boltz.CrudStore),
 	}
 
+	externalStores.EventualEventer = internalStores.EventualEventer
+
 	// The Index store is used for querying indexes. It's a convenient store with only a single value (id), which
 	// is only ever queried using an index set cursor
 	externalStores.Index = boltz.NewBaseStore("invalid", func(id string) error {

controller/persistence/testing.go

@@ -60,6 +60,7 @@ type TestContext struct {
 	fabricStores *db.Stores
 	stores       *Stores
 	controllers  *network.Controllers
+	closeNotify  chan struct{}
 }
 
 func NewTestContext(t *testing.T) *TestContext {
@@ -67,11 +68,17 @@ func NewTestContext(t *testing.T) *TestContext {
 
 	result := &TestContext{
 		BaseTestContext: *boltz.NewTestContext(t),
+		closeNotify:     make(chan struct{}, 1),
 	}
 	result.Impl = result
 	return result
 }
 
+func (ctx *TestContext) Cleanup() {
+	close(ctx.closeNotify)
+	ctx.BaseTestContext.Cleanup()
+}
+
 func (ctx *TestContext) GetStores() *Stores {
 	return ctx.stores
 }
@@ -116,6 +123,9 @@ func (ctx *TestContext) InitWithDbFile(path string) {
 	ctx.NoError(err)
 
 	ctx.NoError(RunMigrations(ctx.GetDb(), ctx.stores))
+
+	ctx.NoError(ctx.stores.EventualEventer.Start(ctx.closeNotify))
+
 }
 
 func (ctx *TestContext) requireNewServicePolicy(policyType PolicyType, identityRoles []string, serviceRoles []string) *ServicePolicy {

controller/server/controller.go

@@ -247,6 +247,10 @@ func (c *Controller) Initialize() {
 
 	}
 
+	if err := c.AppEnv.GetStores().EventualEventer.Start(c.AppEnv.GetHostController().GetCloseNotifyChannel()); err != nil {
+		log.WithError(err).Panic("could not start EventualEventer")
+	}
+
 	c.initialized = true
 }
 

tests/api_error_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*
@@ -89,7 +90,6 @@ func Test_Api_Errors(t *testing.T) {
 
 		resp, err := ctx.newAnonymousClientApiRequest().
 			SetHeader("accept", "appliaction/x-pem-file, application/json").
-
 			Post("enroll?token=" + madeUpToken)
 
 		ctx.Req.NoError(err)

tests/api_session_certificates_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*
@@ -206,4 +207,3 @@ func generateCsr() ([]byte, crypto.PrivateKey, error) {
 
 	return csr, privateKey, nil
 }
-

tests/authenticator_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*
@@ -436,7 +437,6 @@ func Test_Authenticators_NonAdminUsingSelfServiceEndpoints(t *testing.T) {
 	certNonAdminUserSession, err := certNonAdminUserAuthenticator.AuthenticateClientApi(ctx)
 	ctx.Req.NoError(err, "expected no error during non-admin cert authentication")
 
-
 	t.Run("can access their authenticators", func(t *testing.T) {
 		req := require.New(t)
 		resp, err := updbNonAdminSession.newAuthenticatedRequest().Get("/current-identity/authenticators")

tests/ca_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 package tests

tests/config_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/config_type_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/current_identity_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/edge_router_policy_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/edge_router_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/enrollment_ca_auto_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/enrollment_router_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*
@@ -371,7 +372,7 @@ func Test_RouterEnrollment(t *testing.T) {
 
 						ch, err := channel2.NewChannel("apitest", channel2.NewClassicDialer(id, ctx.ControllerConfig.Ctrl.Listener, nil), nil)
 
-						defer func(){
+						defer func() {
 							if ch != nil {
 								_ = ch.Close()
 							}
@@ -383,7 +384,7 @@ func Test_RouterEnrollment(t *testing.T) {
 
 					t.Run("requesting enrollment extension with", func(t *testing.T) {
 						ctx.testContextChanged(t)
-						
+
 						t.Run("no existing router client certificate fails", func(t *testing.T) {
 							ctx.testContextChanged(t)
 							body := gabs.New()
@@ -534,7 +535,6 @@ func Test_RouterEnrollment(t *testing.T) {
 								extensionClientCerts[0].NotBefore.Before(time.Now())
 							})
 
-
 							t.Run("the new server cert has its NotAfter date increased", func(t *testing.T) {
 								ctx.testContextChanged(t)
 								extensionServerCert[0].NotAfter.After(serverCerts[0].NotAfter)
@@ -553,7 +553,7 @@ func Test_RouterEnrollment(t *testing.T) {
 
 								ch, err := channel2.NewChannel("apitest", channel2.NewClassicDialer(id, ctx.ControllerConfig.Ctrl.Listener, nil), nil)
 
-								defer func(){
+								defer func() {
 									if ch != nil {
 										_ = ch.Close()
 									}
@@ -570,7 +570,7 @@ func Test_RouterEnrollment(t *testing.T) {
 
 								ch, err := channel2.NewChannel("apitestextension", channel2.NewClassicDialer(id, ctx.ControllerConfig.Ctrl.Listener, nil), nil)
 
-								defer func(){
+								defer func() {
 									if ch != nil {
 										_ = ch.Close()
 									}

tests/enrollment_updb_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*
@@ -47,7 +48,7 @@ func Test_UpdbEnrollment(t *testing.T) {
 			"updb": updbUsername,
 		}, "enrollment")
 		updbCreate.Set(false, "isAdmin")
-		
+
 		resp, err := ctx.AdminManagementSession.newAuthenticatedRequest().SetBody(updbCreate.String()).Post("identities")
 		ctx.Req.NoError(err)
 

tests/identity_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/mfa_ziti_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/posture_check_domain_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*
@@ -127,7 +128,6 @@ func Test_PostureChecks_Domain(t *testing.T) {
 			ctx.Req.True(enrolledIdentitySession.isServiceVisibleToUser(service.Id))
 		})
 
-
 		t.Run("service has the posture check in its queries", func(t *testing.T) {
 			ctx.testContextChanged(t)
 			code, body := enrolledIdentitySession.query("/services/" + service.Id)
@@ -144,7 +144,7 @@ func Test_PostureChecks_Domain(t *testing.T) {
 			if querySet[0].Path("policyId").Data().(string) == dialPolicy.id {
 				dialSet = querySet[0]
 				bindSet = querySet[1]
-			}  else {
+			} else {
 				dialSet = querySet[1]
 				bindSet = querySet[0]
 			}
@@ -159,7 +159,6 @@ func Test_PostureChecks_Domain(t *testing.T) {
 			ctx.Req.Equal(postureCheck.id, postureQueries[0].Path("id").Data().(string))
 			ctx.Req.Equal(postureCheck.typeId, postureQueries[0].Path("queryType").Data().(string))
 
-
 			t.Run("query is currently failing", func(t *testing.T) {
 				ctx.testContextChanged(t)
 				ctx.Req.False(dialSet.Path("isPassing").Data().(bool))
@@ -205,7 +204,7 @@ func Test_PostureChecks_Domain(t *testing.T) {
 				if querySet[0].Path("policyId").Data().(string) == dialPolicy.id {
 					dialSet = querySet[0]
 					bindSet = querySet[1]
-				}  else {
+				} else {
 					dialSet = querySet[1]
 					bindSet = querySet[0]
 				}
@@ -264,7 +263,7 @@ func Test_PostureChecks_Domain(t *testing.T) {
 					if querySet[0].Path("policyId").Data().(string) == dialPolicy.id {
 						dialSet = querySet[0]
 						bindSet = querySet[1]
-					}  else {
+					} else {
 						dialSet = querySet[1]
 						bindSet = querySet[0]
 					}
@@ -328,7 +327,7 @@ func Test_PostureChecks_Domain(t *testing.T) {
 				if querySet[0].Path("policyId").Data().(string) == dialPolicy.id {
 					dialSet = querySet[0]
 					bindSet = querySet[1]
-				}  else {
+				} else {
 					dialSet = querySet[1]
 					bindSet = querySet[0]
 				}
@@ -395,7 +394,6 @@ func Test_PostureChecks_Domain(t *testing.T) {
 			ctx.Req.True(enrolledIdentitySession.isServiceVisibleToUser(service.Id))
 		})
 
-
 		t.Run("can create session with failing queries via one service policy by having another service policy with no checks", func(t *testing.T) {
 			ctx.testContextChanged(t)
 			resp, err := enrolledIdentitySession.createNewSession(service.Id)

tests/posture_check_mfa_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/posture_check_os_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/posture_check_process_multi_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/posture_check_session_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*
@@ -166,4 +167,4 @@ func Test_PostureChecks_Sessions(t *testing.T) {
 	})
 
 	testServer.waitForDone(ctx, 5*time.Second)
-}
+}

tests/service_edge_router_policy_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/service_list_refresh_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 package tests

tests/service_policy_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/service_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/specs_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*

tests/transit_router_test.go

@@ -1,3 +1,4 @@
+//go:build apitests
 // +build apitests
 
 /*