diff --git a/cmd/icingadb/main.go b/cmd/icingadb/main.go
index 1cfc64a67..65d4d6c62 100644
--- a/cmd/icingadb/main.go
+++ b/cmd/icingadb/main.go
@@ -169,32 +169,34 @@ func run() int {
 	sig := make(chan os.Signal, 1)
 	signal.Notify(sig, os.Interrupt, syscall.SIGTERM, syscall.SIGHUP)
 
-	{
-		var extraStages map[string]history.StageFunc
-		if cfg := cmd.Config.Notifications; cfg.Url != "" {
-			logger.Info("Starting Icinga Notifications source")
-
-			notificationsSource, err := notifications.NewNotificationsClient(
-				ctx,
-				db,
-				rc,
-				logs.GetChildLogger("notifications"),
-				cfg)
-			if err != nil {
-				logger.Fatalw("Can't create Icinga Notifications client from config", zap.Error(err))
-			}
-
-			extraStages = notificationsSource.SyncExtraStages()
+	var notificationsSource *notifications.Client
+	if cfg := cmd.Config.Notifications; cfg.Url != "" {
+		logger.Info("Starting Icinga Notifications source")
+
+		notificationsSource, err = notifications.NewNotificationsClient(
+			db,
+			rc,
+			logs.GetChildLogger("notifications"),
+			cfg)
+		if err != nil {
+			logger.Fatalw("Can't create Icinga Notifications client from config", zap.Error(err))
 		}
+	}
 
-		go func() {
-			logger.Info("Starting history sync")
+	go func() {
+		logger.Info("Starting history sync")
 
-			if err := hs.Sync(ctx, extraStages); err != nil && !utils.IsContextCanceled(err) {
-				logger.Fatalf("%+v", err)
-			}
-		}()
-	}
+		var extraStages map[string]history.StageFunc
+		//if notificationsSource != nil {
+		// We don't need to subscribe to the history pipelines anymore, so disabling them temporarily until
+		// someone works on implementing https://github.com/Icinga/icinga-notifications/issues/409.
+		//extraStages = notificationsSource.SyncExtraStages(ctx)
+		//}
+
+		if err := hs.Sync(ctx, extraStages); err != nil && !utils.IsContextCanceled(err) {
+			logger.Fatalf("%+v", err)
+		}
+	}()
 
 	// Main loop
 	for {
@@ -325,7 +327,7 @@ func run() int {
 
 							logger.Info("Starting config runtime updates sync")
 
-							return rt.Sync(synctx, v1.ConfigFactories, runtimeConfigUpdateStreams, false)
+							return rt.Sync(synctx, v1.ConfigFactories, runtimeConfigUpdateStreams)
 						})
 
 						g.Go(func() error {
@@ -337,7 +339,12 @@ func run() int {
 
 							logger.Info("Starting state runtime updates sync")
 
-							return rt.Sync(synctx, v1.StateFactories, runtimeStateUpdateStreams, true)
+							runtimeUpdatesOpts := []icingadb.RUOption{icingadb.WithAllowParallel()}
+							if notificationsSource != nil {
+								runtimeUpdatesOpts = append(runtimeUpdatesOpts, icingadb.WithRUUpsert(notificationsSource.Submit))
+							}
+
+							return rt.Sync(synctx, v1.StateFactories, runtimeStateUpdateStreams, runtimeUpdatesOpts...)
 						})
 
 						g.Go(func() error {
diff --git a/go.mod b/go.mod
index 10f3a9288..bfc608f91 100644
--- a/go.mod
+++ b/go.mod
@@ -7,7 +7,7 @@ require (
 	github.com/goccy/go-yaml v1.13.0
 	github.com/google/go-cmp v0.7.0
 	github.com/google/uuid v1.6.0
-	github.com/icinga/icinga-go-library v0.9.1-0.20260623130323-faef23cbd4c9
+	github.com/icinga/icinga-go-library v0.9.1-0.20260630080726-15d47aff49f8
 	github.com/jessevdk/go-flags v1.6.1
 	github.com/jmoiron/sqlx v1.4.0
 	github.com/mattn/go-sqlite3 v1.14.47
diff --git a/go.sum b/go.sum
index b032bdd94..76cb0e123 100644
--- a/go.sum
+++ b/go.sum
@@ -36,8 +36,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/icinga/icinga-go-library v0.9.1-0.20260623130323-faef23cbd4c9 h1:HyeACIYSpROzA9bUghELjzsCwnHzM48M6oD2zgKMlfA=
-github.com/icinga/icinga-go-library v0.9.1-0.20260623130323-faef23cbd4c9/go.mod h1:XFsuZEn2hrE3YXQ3NHqnI2UkjdZgJJWb/7D/+VH6MsE=
+github.com/icinga/icinga-go-library v0.9.1-0.20260630080726-15d47aff49f8 h1:YZ+N5qKN9tB6JPUno3oVcVE9TAayAIOZAG8hmVkgYBQ=
+github.com/icinga/icinga-go-library v0.9.1-0.20260630080726-15d47aff49f8/go.mod h1:XFsuZEn2hrE3YXQ3NHqnI2UkjdZgJJWb/7D/+VH6MsE=
 github.com/jessevdk/go-flags v1.6.1 h1:Cvu5U8UGrLay1rZfv/zP7iLpSHGUZ/Ou68T0iX1bBK4=
 github.com/jessevdk/go-flags v1.6.1/go.mod h1:Mk8T1hIAWpOiJiHa9rJASDK2UGWji0EuPGBnNLMooyc=
 github.com/jmoiron/sqlx v1.4.0 h1:1PLqN7S1UYp5t4SrVVnt4nUVNemrDAtxlulVe+Qgm3o=
diff --git a/pkg/icingadb/runtime_updates.go b/pkg/icingadb/runtime_updates.go
index b55fb626d..866fb803c 100644
--- a/pkg/icingadb/runtime_updates.go
+++ b/pkg/icingadb/runtime_updates.go
@@ -10,17 +10,16 @@ import (
 	"github.com/icinga/icinga-go-library/redis"
 	"github.com/icinga/icinga-go-library/strcase"
 	"github.com/icinga/icinga-go-library/structify"
+	"github.com/icinga/icinga-go-library/types"
 	"github.com/icinga/icingadb/pkg/common"
 	"github.com/icinga/icingadb/pkg/contracts"
 	v1 "github.com/icinga/icingadb/pkg/icingadb/v1"
 	"github.com/icinga/icingadb/pkg/icingaredis/telemetry"
 	"github.com/pkg/errors"
-	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 	"golang.org/x/sync/semaphore"
+	"maps"
 	"reflect"
-	"strconv"
-	"strings"
 	"sync"
 )
 
@@ -57,50 +56,193 @@ func (r *RuntimeUpdates) ClearStreams(ctx context.Context) (config, state redis.
 	return
 }
 
-// Sync synchronizes runtime update streams from s.redis to s.db and deletes the original data on success.
-// Note that Sync must be only be called configuration synchronization has been completed.
-// allowParallel allows synchronizing out of order (not FIFO).
+// prepareCustomVarsForSync prepares the channels and goroutines for synchronizing custom variables.
+//
+// The returned channel is the one to which the Redis stream messages for custom variables will be sent.
+func (r *RuntimeUpdates) prepareCustomVarsForSync(ctx context.Context, g *errgroup.Group) chan<- redis.XMessage {
+	updateMessages := make(chan redis.XMessage, r.redis.Options.XReadCount)
+	upsertEntities := make(chan database.Entity, r.redis.Options.XReadCount)
+	deleteIds := make(chan any, r.redis.Options.XReadCount)
+
+	cv := common.NewSyncSubject(v1.NewCustomvar)
+	cvFlat := common.NewSyncSubject(v1.NewCustomvarFlat)
+
+	r.logger.Debug("Syncing runtime updates of " + cv.Name())
+	r.logger.Debug("Syncing runtime updates of " + cvFlat.Name())
+
+	g.Go(structifyStream(
+		ctx, updateMessages, upsertEntities, deleteIds, nil,
+		structify.MakeMapStructifier(
+			reflect.TypeOf(cv.Entity()).Elem(),
+			"json",
+			contracts.SafeInit),
+	))
+
+	customvars, flatCustomvars, errs := v1.ExpandCustomvars(ctx, upsertEntities)
+	com.ErrgroupReceive(g, errs)
+
+	syncableCvs := map[*common.SyncSubject]<-chan database.Entity{cv: customvars, cvFlat: flatCustomvars}
+	for s, cvInCh := range syncableCvs {
+		g.Go(func() error {
+			var counter com.Counter
+			defer periodic.Start(ctx, r.logger.Interval(), func(_ periodic.Tick) {
+				if count := counter.Reset(); count > 0 {
+					r.logger.Infof("Upserted %d %s items", count, s.Name())
+				}
+			}).Stop()
+
+			// Updates must be executed in order, ensure this by using a semaphore with maximum 1.
+			sem := semaphore.NewWeighted(1)
+
+			stmt, placeholders := r.db.BuildUpsertStmt(s.Entity())
+			return r.db.NamedBulkExec(
+				ctx, stmt, r.db.BatchSizeByPlaceholders(placeholders), sem, cvInCh,
+				database.SplitOnDupId[database.Entity],
+				database.OnSuccessIncrement[database.Entity](&counter),
+				database.OnSuccessIncrement[database.Entity](&telemetry.Stats.Config),
+			)
+		})
+	}
+
+	g.Go(func() error {
+		var once sync.Once
+		for {
+			select {
+			case _, ok := <-deleteIds:
+				if !ok {
+					return nil
+				}
+				// Icinga 2 does not send custom var delete events.
+				once.Do(func() {
+					r.logger.DPanic("received unexpected custom var delete event")
+				})
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
+	})
+
+	return updateMessages
+}
+
+// Sync synchronizes runtime update streams from r.redis to r.db and deletes the original data on success.
+//
+// The options parameter can be used to specify additional options for the synchronization, such as allowing
+// parallel execution of updates for the same entity type (bulk DB upsert and delete ops) or providing a callback
+// that is called for each Redis stream message of type "upsert" in parallel with the main database upsert operations
+// for the same entity type.
+//
+// Note that Sync must only be called after configuration synchronization has been completed.
 func (r *RuntimeUpdates) Sync(
-	ctx context.Context, factoryFuncs []database.EntityFactoryFunc, streams redis.Streams, allowParallel bool,
+	ctx context.Context,
+	factoryFuncs []database.EntityFactoryFunc,
+	streams redis.Streams,
+	options ...RUOption,
 ) error {
-	g, ctx := errgroup.WithContext(ctx)
+	if len(streams) != 1 {
+		return errors.New("streams must contain exactly one stream key for the runtime updates")
+	}
 
-	updateMessagesByKey := make(map[string]chan<- redis.XMessage)
+	type messageByKey map[string]chan<- redis.XMessage
+	// xReads is a slice of length 2, where the first element is the map of each entity keys to channels for the
+	// main sync, and the second element is the map for the additional synchronization with custom onUpsert callback
+	// (if specified). This separation is necessary because we want to start a separate xRead goroutine for each sync,
+	// and each xRead goroutine needs its own set of channels to send the messages to.
+	xReads := make([]messageByKey, 2)
 
-	for _, factoryFunc := range factoryFuncs {
-		s := common.NewSyncSubject(factoryFunc)
-		stat := getCounterForEntity(s.Entity())
+	opts := new(RUOptions)
+	for _, opt := range options {
+		opt(opts)
+	}
 
-		updateMessages := make(chan redis.XMessage, r.redis.Options.XReadCount)
-		upsertEntities := make(chan database.Entity, r.redis.Options.XReadCount)
-		deleteIds := make(chan any, r.redis.Options.XReadCount)
-
-		var upsertedFifo chan database.Entity
-		var deletedFifo chan any
-		var upsertCount int
-		var deleteCount int
-		upsertStmt, upsertPlaceholders := r.db.BuildUpsertStmt(s.Entity())
-		if !allowParallel {
-			upsertedFifo = make(chan database.Entity, 1)
-			deletedFifo = make(chan any, 1)
-			upsertCount = 1
-			deleteCount = 1
+	g, ctx := errgroup.WithContext(ctx)
+	prepareForSync := func(s *common.SyncSubject, serializerCh <-chan any) (chan<- redis.XMessage, <-chan database.Entity, <-chan any) {
+		var upsertEntities chan database.Entity
+		var updateMessages chan redis.XMessage
+		var deleteIds chan any
+
+		// Don't use buffered channels if a serializer channel is provided, because the structifyStream goroutine will
+		// block on the serializer channel after each message dispatch to the respective upsertEntities or deleteIds
+		// channel. So, we don't need to unnecessarily use huge buffers in that case!
+		if serializerCh == nil {
+			updateMessages = make(chan redis.XMessage, r.redis.Options.XReadCount)
+			upsertEntities = make(chan database.Entity, r.redis.Options.XReadCount)
+			deleteIds = make(chan any, r.redis.Options.XReadCount)
 		} else {
-			upsertCount = r.db.BatchSizeByPlaceholders(upsertPlaceholders)
-			deleteCount = r.db.Options.MaxPlaceholdersPerStatement
+			updateMessages = make(chan redis.XMessage)
+			upsertEntities = make(chan database.Entity)
+			deleteIds = make(chan any)
 		}
 
-		updateMessagesByKey[fmt.Sprintf("icinga:%s", strcase.Delimited(s.Name(), ':'))] = updateMessages
-
-		r.logger.Debugf("Syncing runtime updates of %s", s.Name())
-
 		g.Go(structifyStream(
-			ctx, updateMessages, upsertEntities, upsertedFifo, deleteIds, deletedFifo,
+			ctx, updateMessages, upsertEntities, deleteIds, serializerCh,
 			structify.MakeMapStructifier(
 				reflect.TypeOf(s.Entity()).Elem(),
 				"json",
 				contracts.SafeInit),
 		))
+		return updateMessages, upsertEntities, deleteIds
+	}
+
+	for _, factoryFunc := range factoryFuncs {
+		s := common.NewSyncSubject(factoryFunc)
+		stat := getCounterForEntity(s.Entity())
+
+		r.logger.Debugf("Syncing runtime updates of %s", s.Name())
+
+		key := fmt.Sprintf("icinga:%s", strcase.Delimited(s.Name(), ':'))
+
+		if opts.upsertFn != nil {
+			r.logger.Debugf("Starting additional sync with custom onUpsert callback for %s", s.Name())
+
+			serializerCh := make(chan any)
+			updateMessages, upsertEntities, deleteIds := prepareForSync(s, serializerCh)
+			if xReads[1] == nil {
+				xReads[1] = make(messageByKey)
+			}
+			xReads[1][key] = updateMessages
+
+			g.Go(func() error {
+				defer close(serializerCh)
+
+				for {
+					select {
+					case <-ctx.Done():
+						return ctx.Err()
+					case entity, ok := <-upsertEntities:
+						if !ok {
+							return nil
+						}
+						if err := opts.upsertFn(ctx, entity); err != nil {
+							return errors.Wrapf(err, "onUpsert callback failed for entity with ID %v", entity.ID())
+						}
+					case _, ok := <-deleteIds:
+						if !ok {
+							return nil
+						}
+						// Nothing to do for deletes in the onUpsert callback, but we need to consume
+						// the channel to avoid blocking the structifyStream goroutine.
+					}
+
+					select {
+					case <-ctx.Done():
+						return ctx.Err()
+					case serializerCh <- struct{}{}:
+					}
+				}
+			})
+		}
+
+		var serializerCh chan any
+		if !opts.allowParallel {
+			serializerCh = make(chan any)
+		}
+
+		updateMessages, upsertEntities, deleteIds := prepareForSync(s, serializerCh)
+		if xReads[0] == nil {
+			xReads[0] = make(messageByKey)
+		}
+		xReads[0][key] = updateMessages
 
 		g.Go(func() error {
 			var counter com.Counter
@@ -116,8 +258,14 @@ func (r *RuntimeUpdates) Sync(
 			onSuccess := []database.OnSuccess[database.Entity]{
 				database.OnSuccessIncrement[database.Entity](&counter), database.OnSuccessIncrement[database.Entity](stat),
 			}
-			if !allowParallel {
-				onSuccess = append(onSuccess, database.OnSuccessSendTo(upsertedFifo))
+
+			var upsertCount int
+			upsertStmt, upsertPlaceholders := r.db.BuildUpsertStmt(s.Entity())
+			if !opts.allowParallel {
+				upsertCount = 1
+				onSuccess = append(onSuccess, database.OnSuccessApplyAndSendTo(serializerCh, func(e database.Entity) any { return e }))
+			} else {
+				upsertCount = r.db.BatchSizeByPlaceholders(upsertPlaceholders)
 			}
 
 			return r.db.NamedBulkExec(
@@ -136,8 +284,12 @@ func (r *RuntimeUpdates) Sync(
 			sem := r.db.GetSemaphoreForTable(database.TableName(s.Entity()))
 
 			onSuccess := []database.OnSuccess[any]{database.OnSuccessIncrement[any](&counter), database.OnSuccessIncrement[any](stat)}
-			if !allowParallel {
-				onSuccess = append(onSuccess, database.OnSuccessSendTo(deletedFifo))
+			var deleteCount int
+			if !opts.allowParallel {
+				deleteCount = 1
+				onSuccess = append(onSuccess, database.OnSuccessSendTo(serializerCh))
+			} else {
+				deleteCount = r.db.Options.MaxPlaceholdersPerStatement
 			}
 
 			return r.db.BulkExec(ctx, r.db.BuildDeleteStmt(s.Entity()), deleteCount, sem, deleteIds, onSuccess...)
@@ -146,97 +298,46 @@ func (r *RuntimeUpdates) Sync(
 
 	// customvar and customvar_flat sync don't need to be processed with state updates too.
 	if _, exists := streams["icinga:runtime"]; exists {
-		updateMessages := make(chan redis.XMessage, r.redis.Options.XReadCount)
-		upsertEntities := make(chan database.Entity, r.redis.Options.XReadCount)
-		deleteIds := make(chan any, r.redis.Options.XReadCount)
-
-		cv := common.NewSyncSubject(v1.NewCustomvar)
-		cvFlat := common.NewSyncSubject(v1.NewCustomvarFlat)
-
-		r.logger.Debug("Syncing runtime updates of " + cv.Name())
-		r.logger.Debug("Syncing runtime updates of " + cvFlat.Name())
-
-		updateMessagesByKey["icinga:"+strcase.Delimited(cv.Name(), ':')] = updateMessages
-		g.Go(structifyStream(
-			ctx, updateMessages, upsertEntities, nil, deleteIds, nil,
-			structify.MakeMapStructifier(
-				reflect.TypeOf(cv.Entity()).Elem(),
-				"json",
-				contracts.SafeInit),
-		))
-
-		customvars, flatCustomvars, errs := v1.ExpandCustomvars(ctx, upsertEntities)
-		com.ErrgroupReceive(g, errs)
-
-		cvStmt, cvPlaceholders := r.db.BuildUpsertStmt(cv.Entity())
-		cvCount := r.db.BatchSizeByPlaceholders(cvPlaceholders)
-		g.Go(func() error {
-			var counter com.Counter
-			defer periodic.Start(ctx, r.logger.Interval(), func(_ periodic.Tick) {
-				if count := counter.Reset(); count > 0 {
-					r.logger.Infof("Upserted %d %s items", count, cv.Name())
-				}
-			}).Stop()
-
-			// Updates must be executed in order, ensure this by using a semaphore with maximum 1.
-			sem := semaphore.NewWeighted(1)
-
-			return r.db.NamedBulkExec(
-				ctx, cvStmt, cvCount, sem, customvars, database.SplitOnDupId[database.Entity],
-				database.OnSuccessIncrement[database.Entity](&counter),
-				database.OnSuccessIncrement[database.Entity](&telemetry.Stats.Config),
-			)
-		})
-
-		cvFlatStmt, cvFlatPlaceholders := r.db.BuildUpsertStmt(cvFlat.Entity())
-		cvFlatCount := r.db.BatchSizeByPlaceholders(cvFlatPlaceholders)
-		g.Go(func() error {
-			var counter com.Counter
-			defer periodic.Start(ctx, r.logger.Interval(), func(_ periodic.Tick) {
-				if count := counter.Reset(); count > 0 {
-					r.logger.Infof("Upserted %d %s items", count, cvFlat.Name())
-				}
-			}).Stop()
-
-			// Updates must be executed in order, ensure this by using a semaphore with maximum 1.
-			sem := semaphore.NewWeighted(1)
+		if xReads[0] == nil {
+			xReads[0] = make(messageByKey)
+		}
+		xReads[0]["icinga:"+strcase.Delimited(types.Name(v1.Customvar{}), ':')] = r.prepareCustomVarsForSync(ctx, g)
+	}
 
-			return r.db.NamedBulkExec(
-				ctx, cvFlatStmt, cvFlatCount, sem, flatCustomvars,
-				database.SplitOnDupId[database.Entity], database.OnSuccessIncrement[database.Entity](&counter),
-				database.OnSuccessIncrement[database.Entity](&telemetry.Stats.Config),
-			)
-		})
+	// Since all xRead goroutines are going to consume messages from the same stream independently, we are only
+	// allowed to send XDel commands after we've successfully dispatched all messages to the corresponding
+	// updateMessages channels. For the database ops, it doesn't really matter when the msgs are acked, but for
+	// the onUpsert callback, the per type updates are processed sequentially, so the xRead will block each time
+	// it tries to send a message to the chOuts channel until the callback has processed the previous one. When
+	// the callback fails to process the previous one (which we already deleted from the stream), we'll either
+	// trigger a HA handover or crash Icinga DB fatally, in which case losing that message is not a big deal.
+	var xRedisMessageAcks []<-chan string
+	for _, chOuts := range xReads {
+		if chOuts == nil {
+			continue
+		}
+		ackMessageCh := make(chan string)
+		xRedisMessageAcks = append(xRedisMessageAcks, ackMessageCh)
 
-		g.Go(func() error {
-			var once sync.Once
-			for {
-				select {
-				case _, ok := <-deleteIds:
-					if !ok {
-						return nil
-					}
-					// Icinga 2 does not send custom var delete events.
-					once.Do(func() {
-						r.logger.DPanic("received unexpected custom var delete event")
-					})
-				case <-ctx.Done():
-					return ctx.Err()
-				}
-			}
-		})
+		g.Go(r.xRead(ctx, chOuts, ackMessageCh, maps.Clone(streams)))
 	}
 
-	g.Go(r.xRead(ctx, updateMessagesByKey, streams))
+	g.Go(func() error { return r.redis.XDelOnAllConsumersAck(ctx, streams.Option()[0], xRedisMessageAcks...) })
 
 	return g.Wait()
 }
 
 // xRead reads from the runtime update streams and sends the data to the corresponding updateMessages channel.
 // The updateMessages channel is determined by a "redis_key" on each redis message.
-func (r *RuntimeUpdates) xRead(ctx context.Context, updateMessagesByKey map[string]chan<- redis.XMessage, streams redis.Streams) func() error {
+func (r *RuntimeUpdates) xRead(
+	ctx context.Context,
+	updateMessagesByKey map[string]chan<- redis.XMessage,
+	acknowledgementOutCh chan<- string,
+	streams redis.Streams,
+) func() error {
 	return func() error {
 		defer func() {
+			close(acknowledgementOutCh)
 			for _, updateMessages := range updateMessagesByKey {
 				close(updateMessages)
 			}
@@ -251,7 +352,6 @@ func (r *RuntimeUpdates) xRead(ctx context.Context, updateMessagesByKey map[stri
 				return errors.Wrap(err, "can't read runtime updates")
 			}
 
-			pipe := r.redis.Pipeline()
 			for _, stream := range rs {
 				var id string
 
@@ -273,25 +373,14 @@ func (r *RuntimeUpdates) xRead(ctx context.Context, updateMessagesByKey map[stri
 					case <-ctx.Done():
 						return ctx.Err()
 					}
-				}
-
-				tsAndSerial := strings.Split(id, "-")
-				if s, err := strconv.ParseUint(tsAndSerial[1], 10, 64); err == nil {
-					tsAndSerial[1] = strconv.FormatUint(s+1, 10)
-				}
 
-				pipe.XTrimMinIDApprox(ctx, stream.Stream, strings.Join(tsAndSerial, "-"), 0)
-				streams[stream.Stream] = id
-			}
-
-			if cmds, err := pipe.Exec(ctx); err != nil {
-				r.logger.Errorw("Can't execute Redis pipeline", zap.Error(errors.WithStack(err)))
-			} else {
-				for _, cmd := range cmds {
-					if cmd.Err() != nil {
-						r.logger.Errorw("Can't trim runtime updates stream", zap.Error(redis.WrapCmdErr(cmd)))
+					select {
+					case acknowledgementOutCh <- message.ID:
+					case <-ctx.Done():
+						return ctx.Err()
 					}
 				}
+				streams[stream.Stream] = id
 			}
 		}
 	}
@@ -302,34 +391,27 @@ func (r *RuntimeUpdates) xRead(ctx context.Context, updateMessagesByKey map[stri
 // Converted entities are inserted into the upsertEntities or deleteIds channel depending on the "runtime_type" message field.
 func structifyStream(
 	ctx context.Context,
-	updateMessages <-chan redis.XMessage,
-	upsertEntities chan<- database.Entity,
-	upserted <-chan database.Entity,
-	deleteIds chan<- any,
-	deleted <-chan any,
+	messagesInCh <-chan redis.XMessage,
+	upsertEntitiesOutCh chan<- database.Entity,
+	deleteIdsOutCh chan<- any,
+	serializerInCh <-chan any,
 	structifier structify.MapStructifier,
 ) func() error {
-	if upserted == nil {
-		ch := make(chan database.Entity)
-		close(ch)
-		upserted = ch
-	}
-
-	if deleted == nil {
+	if serializerInCh == nil {
 		ch := make(chan any)
 		close(ch)
-		deleted = ch
+		serializerInCh = ch
 	}
 
 	return func() error {
 		defer func() {
-			close(upsertEntities)
-			close(deleteIds)
+			close(upsertEntitiesOutCh)
+			close(deleteIdsOutCh)
 		}()
 
 		for {
 			select {
-			case message, ok := <-updateMessages:
+			case message, ok := <-messagesInCh:
 				if !ok {
 					return nil
 				}
@@ -351,34 +433,56 @@ func structifyStream(
 
 				if runtimeType == "upsert" {
 					select {
-					case upsertEntities <- entity:
-					case <-ctx.Done():
-						return ctx.Err()
-					}
-
-					select {
-					case <-upserted:
+					case upsertEntitiesOutCh <- entity:
 					case <-ctx.Done():
 						return ctx.Err()
 					}
 				} else if runtimeType == "delete" {
 					select {
-					case deleteIds <- entity.ID():
-					case <-ctx.Done():
-						return ctx.Err()
-					}
-
-					select {
-					case <-deleted:
+					case deleteIdsOutCh <- entity.ID():
 					case <-ctx.Done():
 						return ctx.Err()
 					}
 				} else {
 					return errors.Errorf("invalid runtime type: %s", runtimeType)
 				}
+
+				select {
+				case <-serializerInCh:
+				case <-ctx.Done():
+					return ctx.Err()
+				}
 			case <-ctx.Done():
 				return ctx.Err()
 			}
 		}
 	}
 }
+
+// RUOption defines an option for the [RuntimeUpdates.Sync] method.
+type RUOption func(*RUOptions)
+
+// RUUpsertFunc defines the type of the callback that can be provided to the [RuntimeUpdates.Sync] method via the [WithRUUpsert] option.
+type RUUpsertFunc func(context.Context, database.Entity) error
+
+// RUOptions defines options for the [RuntimeUpdates.Sync] method.
+type RUOptions struct {
+	allowParallel bool
+	upsertFn      RUUpsertFunc
+}
+
+// WithAllowParallel allows parallel execution of runtime updates for the same entity type.
+func WithAllowParallel() RUOption {
+	return func(opts *RUOptions) { opts.allowParallel = true }
+}
+
+// WithRUUpsert allows providing a callback that is called for each Redis stream message of type "upsert".
+//
+// If this option is used, [RuntimeUpdates.Sync] will start a separate xRead goroutine for the corresponding stream,
+// thus allowing the provided callback to run in parallel with the main database upsert operations for the same entity
+// type. Note that the callback is called for each message of type "upsert", regardless of the entity type, so it is
+// the responsibility of that callback to filter the entities if necessary. Also note that the callback must be safe
+// for concurrent execution, as it may be called in parallel with different types of entities.
+func WithRUUpsert(fn RUUpsertFunc) RUOption {
+	return func(opts *RUOptions) { opts.upsertFn = fn }
+}
diff --git a/pkg/notifications/fetch.go b/pkg/notifications/fetch.go
index cfb63ee9e..f2730983b 100644
--- a/pkg/notifications/fetch.go
+++ b/pkg/notifications/fetch.go
@@ -71,6 +71,8 @@ func (client *Client) fetchObjectsFromRedis(
 								"cannot JSON unmarshal Redis HMGET result for %q with key %q",
 								key, pair.Field)
 						}
+						obj.isVolatile = obj.IsVolatile.Valid && obj.IsVolatile.Bool
+						obj.IsVolatile = types.Bool{} // Reset it, so that it is omitted from the JSON output.
 						out[pair.Field] = obj
 					}
 				}
@@ -251,7 +253,9 @@ type icingaObject struct {
 	Name        string `json:"name" db:"name"`
 	DisplayName string `json:"display_name" db:"display_name"`
 	// Vars might be unpopulated for certain types, such as host or service groups.
-	Vars map[string]any `json:"vars,omitempty" db:"-"`
+	Vars       map[string]any `json:"vars,omitempty" db:"-"`
+	IsVolatile types.Bool     `json:"is_volatile,omitzero" db:"-"`
+	isVolatile bool           // internal representation of IsVolatile, so that it can be omitted from generated JSON
 }
 
 // relations of an object to be passed to Icinga Notifications.
diff --git a/pkg/notifications/notifications.go b/pkg/notifications/notifications.go
index 52b000865..69a11dfc7 100644
--- a/pkg/notifications/notifications.go
+++ b/pkg/notifications/notifications.go
@@ -6,11 +6,13 @@ import (
 	"reflect"
 	"time"
 
+	"github.com/icinga/icinga-go-library/backoff"
 	"github.com/icinga/icinga-go-library/database"
 	"github.com/icinga/icinga-go-library/logging"
 	"github.com/icinga/icinga-go-library/notifications/event"
 	"github.com/icinga/icinga-go-library/notifications/source"
 	"github.com/icinga/icinga-go-library/redis"
+	"github.com/icinga/icinga-go-library/retry"
 	"github.com/icinga/icinga-go-library/structify"
 	"github.com/icinga/icinga-go-library/types"
 	"github.com/icinga/icinga-go-library/utils"
@@ -18,6 +20,7 @@ import (
 	"github.com/icinga/icingadb/pkg/common"
 	"github.com/icinga/icingadb/pkg/contracts"
 	"github.com/icinga/icingadb/pkg/icingadb/history"
+	v1 "github.com/icinga/icingadb/pkg/icingadb/v1"
 	v1history "github.com/icinga/icingadb/pkg/icingadb/v1/history"
 	"github.com/icinga/icingadb/pkg/icingaredis/telemetry"
 	"github.com/pkg/errors"
@@ -34,9 +37,6 @@ type fetchableEvent struct {
 //
 // This method can be called with a nil slice to populate the event.Event without any fetching.
 func (ev *fetchableEvent) completeAndUpdate(ctx context.Context, attributes []string) error {
-	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
-	defer cancel()
-
 	for _, attribute := range attributes {
 		err := ev.relations.complete(ctx, attribute)
 		if err != nil {
@@ -62,15 +62,12 @@ type Client struct {
 	db     *database.DB
 	logger *logging.Logger
 
-	ctx context.Context
-
 	notificationsClient *source.Client // The Icinga Notifications client used to interact with the API.
 	redisClient         *redis.Client  // redisClient is the Redis client used to fetch host and service names for events.
 }
 
 // NewNotificationsClient creates a new Client connected to an existing database and logger.
 func NewNotificationsClient(
-	ctx context.Context,
 	db *database.DB,
 	rc *redis.Client,
 	logger *logging.Logger,
@@ -87,8 +84,6 @@ func NewNotificationsClient(
 		db:     db,
 		logger: logger,
 
-		ctx: ctx,
-
 		notificationsClient: notificationsClient,
 		redisClient:         rc,
 	}, nil
@@ -143,36 +138,51 @@ func (client *Client) buildCommonEvent(
 		}
 	}
 
-	ev := &fetchableEvent{
+	return &fetchableEvent{
 		Event: &event.Event{
 			Name: objectName,
 			URL:  objectUrl,
 			Tags: objectTags,
 		},
 		relations: rel,
-	}
-
-	if err = ev.completeAndUpdate(ctx, client.DefaultRelations); err != nil {
-		return nil, errors.Wrap(err, "cannot complete event relations")
-	}
-
-	return ev, nil
+	}, nil
 }
 
-// buildStateHistoryEvent builds a fully initialized event.Event from a state history entry.
+// errNonVolatileNonHardState is returned when a non-hard state change is attempted to be submitted for a non-volatile checkable.
+var errNonVolatileNonHardState = errors.New("non-hard state change for non-volatile checkable")
+
+// buildStateEvent builds a fully initialized event.Event from a state history entry.
 //
 // The resulted event will have all the necessary information for a state change event, and must
 // not be further modified by the caller.
-func (client *Client) buildStateHistoryEvent(ctx context.Context, h *v1history.StateHistory) (*fetchableEvent, error) {
-	ev, err := client.buildCommonEvent(ctx, h.HostId, h.ServiceId)
+func (client *Client) buildStateEvent(ctx context.Context, s *v1.State, hostId, serviceId types.Binary) (*fetchableEvent, error) {
+	ev, err := client.buildCommonEvent(ctx, hostId, serviceId)
 	if err != nil {
-		return nil, errors.Wrapf(err, "cannot build event for %q,%q", h.HostId, h.ServiceId)
+		return nil, errors.Wrapf(err, "cannot build event for %q,%q", hostId, serviceId)
 	}
 
-	ev.Type = event.TypeState
+	if s.Output.Valid {
+		ev.Message = s.Output.String
+	}
+	if s.LongOutput.Valid {
+		ev.Message += "\n" + s.LongOutput.String
+	}
 
-	if h.ServiceId != nil {
-		switch h.HardState {
+	var isVolatile bool
+	if serviceId != nil {
+		isVolatile = ev.Services[0].isVolatile
+	} else {
+		isVolatile = ev.Host.isVolatile
+	}
+
+	// If the checkable is volatile, it's always treated as a hard state change, but `StateType` is still set
+	// to `SOFT` due to an Icinga 2 bug (see https://github.com/Icinga/icinga2/issues/10879).
+	if s.StateType != common.HardState && !isVolatile {
+		return nil, errNonVolatileNonHardState
+	}
+
+	if serviceId != nil {
+		switch s.HardState {
 		case 0:
 			ev.Severity = event.SeverityOK
 		case 1:
@@ -182,24 +192,51 @@ func (client *Client) buildStateHistoryEvent(ctx context.Context, h *v1history.S
 		case 3:
 			ev.Severity = event.SeverityErr
 		default:
-			return nil, fmt.Errorf("unexpected service state %d", h.HardState)
+			return nil, fmt.Errorf("unexpected service state %d", s.HardState)
 		}
 	} else {
-		switch h.HardState {
+		switch s.HardState {
 		case 0:
 			ev.Severity = event.SeverityOK
 		case 1:
 			ev.Severity = event.SeverityCrit
 		default:
-			return nil, fmt.Errorf("unexpected host state %d", h.HardState)
+			return nil, fmt.Errorf("unexpected host state %d", s.HardState)
 		}
 	}
 
-	if h.Output.Valid {
-		ev.Message = h.Output.String
+	inDowntime := s.InDowntime.Valid && s.InDowntime.Bool
+	isAcked := s.IsAcknowledged.Valid && s.IsAcknowledged.Bool
+	isFlapping := s.IsFlapping.Valid && s.IsFlapping.Bool
+	ev.Muted = types.MakeBool(inDowntime || isAcked || isFlapping)
+	if ev.IsMuted() {
+		ev.MutedReason = "Checkable is muted due to"
+		if inDowntime {
+			ev.MutedReason += " currently active downtime"
+		}
+		if isAcked && inDowntime {
+			ev.MutedReason += ", and an acknowledgement"
+		} else if isAcked {
+			ev.MutedReason += " an acknowledgement"
+		}
+		if isFlapping && (inDowntime || isAcked) {
+			ev.MutedReason += ", and flapping as well"
+		} else if isFlapping {
+			ev.MutedReason += " flapping state"
+		}
+		ev.MutedReason += "."
+	} else {
+		ev.MutedReason = "Checkable is not muted (no active downtime, no acknowledgement, and not flapping)"
 	}
-	if h.LongOutput.Valid {
-		ev.Message += "\n" + h.LongOutput.String
+
+	ev.Incident = types.MakeBool(true)
+	if ev.Severity == event.SeverityOK && !ev.IsMuted() {
+		// If the object is still muted, we don't close incidents even with OK state changes.
+		// See https://github.com/Icinga/icingadb/issues/1127#issuecomment-4691435590 for details.
+		ev.Close = types.MakeBool(true)
+	} else if s.PreviousHardState == s.HardState {
+		// NON-OK hard state changes that do not change the state are volatile ones, so set the notify flag.
+		ev.Notify = types.MakeBool(true)
 	}
 
 	return ev, nil
@@ -207,6 +244,8 @@ func (client *Client) buildStateHistoryEvent(ctx context.Context, h *v1history.S
 
 // buildDowntimeHistoryMetaEvent from a downtime history entry.
 func (client *Client) buildDowntimeHistoryMetaEvent(ctx context.Context, h *v1history.DowntimeHistoryMeta) (*fetchableEvent, error) {
+	defer func() { panic("downtime history event generation is incomplete and not yet implemented") }()
+
 	ev, err := client.buildCommonEvent(ctx, h.HostId, h.ServiceId)
 	if err != nil {
 		return nil, errors.Wrapf(err, "cannot build event for %q,%q", h.HostId, h.ServiceId)
@@ -214,23 +253,15 @@ func (client *Client) buildDowntimeHistoryMetaEvent(ctx context.Context, h *v1hi
 
 	switch h.EventType {
 	case "downtime_start":
-		ev.Type = event.TypeDowntimeStart
-		ev.Username = h.Author
 		ev.Message = h.Comment
-		ev.Mute = types.MakeBool(true)
-		ev.MuteReason = "Checkable is in downtime"
 
 	case "downtime_end":
-		ev.Mute = types.MakeBool(false)
 		if h.HasBeenCancelled.Valid && h.HasBeenCancelled.Bool {
-			ev.Type = event.TypeDowntimeRemoved
 			ev.Message = "Downtime was cancelled"
-
 			if h.CancelledBy.Valid {
-				ev.Username = h.CancelledBy.String
+				ev.Message += " (cancelled by " + h.CancelledBy.String + ")"
 			}
 		} else {
-			ev.Type = event.TypeDowntimeEnd
 			ev.Message = "Downtime expired"
 		}
 
@@ -243,24 +274,21 @@ func (client *Client) buildDowntimeHistoryMetaEvent(ctx context.Context, h *v1hi
 
 // buildFlappingHistoryEvent from a flapping history entry.
 func (client *Client) buildFlappingHistoryEvent(ctx context.Context, h *v1history.FlappingHistory) (*fetchableEvent, error) {
+	defer func() { panic("flapping history event generation is incomplete and not yet implemented") }()
+
 	ev, err := client.buildCommonEvent(ctx, h.HostId, h.ServiceId)
 	if err != nil {
 		return nil, errors.Wrapf(err, "cannot build event for %q,%q", h.HostId, h.ServiceId)
 	}
 
 	if h.PercentStateChangeEnd.Valid {
-		ev.Type = event.TypeFlappingEnd
 		ev.Message = fmt.Sprintf(
 			"Checkable stopped flapping (Current flapping value %.2f%% < low threshold %.2f%%)",
 			h.PercentStateChangeEnd.Float64, h.FlappingThresholdLow)
-		ev.Mute = types.MakeBool(false)
 	} else if h.PercentStateChangeStart.Valid {
-		ev.Type = event.TypeFlappingStart
 		ev.Message = fmt.Sprintf(
 			"Checkable started flapping (Current flapping value %.2f%% > high threshold %.2f%%)",
 			h.PercentStateChangeStart.Float64, h.FlappingThresholdHigh)
-		ev.Mute = types.MakeBool(true)
-		ev.MuteReason = "Checkable is flapping"
 	} else {
 		return nil, errors.New("flapping history entry has neither percent_state_change_start nor percent_state_change_end")
 	}
@@ -270,33 +298,24 @@ func (client *Client) buildFlappingHistoryEvent(ctx context.Context, h *v1histor
 
 // buildAcknowledgementHistoryEvent from an acknowledgment history entry.
 func (client *Client) buildAcknowledgementHistoryEvent(ctx context.Context, h *v1history.AcknowledgementHistory) (*fetchableEvent, error) {
+	defer func() { panic("acknowledgement history event generation is incomplete and not yet implemented") }()
+
 	ev, err := client.buildCommonEvent(ctx, h.HostId, h.ServiceId)
 	if err != nil {
 		return nil, errors.Wrapf(err, "cannot build event for %q,%q", h.HostId, h.ServiceId)
 	}
 
 	if !h.ClearTime.Time().IsZero() {
-		ev.Type = event.TypeAcknowledgementCleared
 		ev.Message = "Acknowledgement was cleared"
-		ev.Mute = types.MakeBool(false)
-
 		if h.ClearedBy.Valid {
-			ev.Username = h.ClearedBy.String
+			ev.Message += " (cleared by " + h.ClearedBy.String + ")"
 		}
 	} else if !h.SetTime.Time().IsZero() {
-		ev.Type = event.TypeAcknowledgementSet
-		ev.Mute = types.MakeBool(true)
-		ev.MuteReason = "Checkable was acknowledged"
-
 		if h.Comment.Valid {
 			ev.Message = h.Comment.String
 		} else {
 			ev.Message = "Checkable was acknowledged"
 		}
-
-		if h.Author.Valid {
-			ev.Username = h.Author.String
-		}
 	} else {
 		return nil, errors.New("acknowledgment history entry has neither a set_time nor a clear_time")
 	}
@@ -307,103 +326,138 @@ func (client *Client) buildAcknowledgementHistoryEvent(ctx context.Context, h *v
 // Submit this [database.Entity] to the Icinga Notifications API.
 //
 // Based on the entity's type, a different kind of event will be constructed. The event will be sent to the API in a
-// blocking fashion.
+// blocking fashion and will be retried with an exponential backoff in case of retryable errors until a non-retryable
+// error occurs (like ctx cancellation) or the deadline is exceeded. In other words, when this method returns an error,
+// then it usually means that there's nothing it can do anymore to successfully submit the event, thus it should be
+// treated as a fatal error.
 //
-// Returns true if this entity was processed or cannot be processed any further. Returns false if this entity should be
-// retried later.
-//
-// This method uses the Client's logger.
-func (client *Client) Submit(entity database.Entity) bool {
-	if client.ctx.Err() != nil {
-		client.logger.Errorw("Cannot process submitted entity as client context is done", zap.Error(client.ctx.Err()))
-		return true
-	}
-
+// Note that this function is used as [icingadb.RUUpsertFunc] for the runtime updates pipeline, so its signature must
+// match the [icingadb.RUUpsertFunc] type.
+func (client *Client) Submit(ctx context.Context, entity database.Entity) error {
 	var (
 		ev       *fetchableEvent
 		eventErr error
 	)
 
-	// Keep the type switch in sync with the values of SyncKeyStructPtrs below.
+	canIgnoreStateUpdate := func(s *v1.State) bool {
+		// Ignore PENDING -> OK, otherwise we'll have a bunch of incidents that are be closed immediately.
+		// Also ignore any Pending states (99), as these are not relevant for notifications.
+		return s.HardState == 99 || (s.HardState == 0 && s.PreviousHardState == 99)
+	}
+
 	switch h := entity.(type) {
 	case *v1history.AcknowledgementHistory:
-		ev, eventErr = client.buildAcknowledgementHistoryEvent(client.ctx, h)
+		ev, eventErr = client.buildAcknowledgementHistoryEvent(ctx, h)
 
 	case *v1history.DowntimeHistoryMeta:
-		ev, eventErr = client.buildDowntimeHistoryMetaEvent(client.ctx, h)
+		ev, eventErr = client.buildDowntimeHistoryMetaEvent(ctx, h)
 
 	case *v1history.FlappingHistory:
-		ev, eventErr = client.buildFlappingHistoryEvent(client.ctx, h)
+		ev, eventErr = client.buildFlappingHistoryEvent(ctx, h)
 
-	case *v1history.StateHistory:
-		if h.StateType != common.HardState {
-			return true
+	case *v1.HostState:
+		if canIgnoreStateUpdate(&h.State) {
+			return nil
+		}
+		ev, eventErr = client.buildStateEvent(ctx, &h.State, h.Id, nil)
+
+	case *v1.ServiceState:
+		if canIgnoreStateUpdate(&h.State) {
+			return nil
 		}
-		ev, eventErr = client.buildStateHistoryEvent(client.ctx, h)
+		ev, eventErr = client.buildStateEvent(ctx, &h.State, h.HostId, h.ServiceId)
+
+	case *v1.DependencyEdgeState, *v1.RedundancygroupState:
+		// Nothing to do here, we only received these because they're part of the runtime state update pipeline.
+		return nil
 
 	default:
-		client.logger.Error("Cannot process unsupported type", zap.String("type", fmt.Sprintf("%T", h)))
-		return true
+		client.logger.Errorw("Cannot process unsupported type", zap.String("type", fmt.Sprintf("%T", h)))
+		return nil
 	}
 
 	if eventErr != nil {
-		client.logger.Errorw("Cannot build event from history entry",
-			zap.String("type", fmt.Sprintf("%T", entity)),
-			zap.Error(eventErr))
-		return true
+		if !errors.Is(eventErr, errNonVolatileNonHardState) {
+			client.logger.Errorw("Cannot build event for entity, skipping submission",
+				zap.String("type", fmt.Sprintf("%T", entity)),
+				zap.Error(eventErr))
+		}
+		return nil
 	} else if ev == nil {
 		// This really should not happen.
 		client.logger.Errorw("No event was built, but no error was reported",
 			zap.String("type", fmt.Sprintf("%T", entity)))
-		return true
+		return nil
 	}
 
-	maxAttempts := 3
-	for attempt := range maxAttempts {
-		attributes, err := client.notificationsClient.ProcessEvent(client.ctx, ev.Event, true)
-		if errors.Is(err, source.ErrAttrsNegotiation) {
-			client.logger.Debugw("Icinga Notifications requested more attributes",
-				zap.String("event", ev.Name),
-				zap.Strings("attributes", attributes))
-
-			if attempt == maxAttempts-1 {
-				// Another completeAndUpdate call would be useless, as it won't be evaluated anymore.
-				break
-			}
+	if err := ev.Validate(); err != nil {
+		client.logger.Errorw("BUG: generated event is invalid, skipping submission",
+			zap.Any("event", ev.Event),
+			zap.Any("entity", entity),
+			zap.Error(err))
+		return nil
+	}
 
-			err := ev.completeAndUpdate(client.ctx, attributes)
-			if err != nil {
-				client.logger.Errorw("Cannot fetch required attribute for event",
+	attributes := client.DefaultRelations
+	return retry.WithBackoff(
+		ctx,
+		func(ctx context.Context) (err error) {
+			for {
+				if err := ev.completeAndUpdate(ctx, attributes); err != nil {
+					client.logger.Errorw("Cannot fetch required attribute for event",
+						zap.String("event", ev.Name),
+						zap.Strings("attributes", attributes),
+						zap.Error(err))
+
+					// ev.completeAndUpdate retries any retryable errors internally, so if we get an error here,
+					// it means that we cannot complete the event with the given attributes, so we should not retry
+					// anymore. Therefore, we return a non-retryable error here instead of propagating the original one.
+					return errors.New("cannot complete event relations")
+				}
+
+				attributes, err = client.notificationsClient.ProcessEvent(ctx, ev.Event, true)
+				if errors.Is(err, source.ErrAttrsNegotiation) {
+					client.logger.Debugw("Icinga Notifications requested more attributes",
+						zap.String("event", ev.Name),
+						zap.Strings("attributes", attributes))
+					continue
+				}
+				return err
+			}
+		},
+		retry.Retryable,
+		backoff.DefaultBackoff,
+		retry.Settings{
+			Timeout: retry.DefaultTimeout,
+			OnSuccess: func(elapsed time.Duration, attempt uint64, lastErr error) {
+				telemetry.Stats.NotificationSync.Add(1)
+
+				client.logger.Debugw("Successfully submitted event to Icinga Notifications",
 					zap.String("event", ev.Name),
-					zap.Strings("attributes", attributes),
+					zap.Uint64("attempt", attempt),
+					zap.Duration("elapsed", elapsed),
+					zap.Error(lastErr))
+			},
+			OnRetryableError: func(elapsed time.Duration, attempt uint64, err, lastErr error) {
+				client.logger.Warnw("Cannot submit event to Icinga Notifications",
+					zap.String("event", ev.Name),
+					zap.Uint64("attempt", attempt),
+					zap.Duration("elapsed", elapsed),
 					zap.Error(err))
-				return false
-			}
-		} else if err != nil {
-			client.logger.Errorw("Cannot submit event to Icinga Notifications",
-				zap.String("event", ev.Name),
-				zap.Error(err))
-			return false
-		} else {
-			client.logger.Debugw("Successfully submitted event to Icinga Notifications", zap.String("event", ev.Name))
-			return true
-		}
-	}
-
-	client.logger.Warnw("Failed to submit event in three attempts", zap.String("event", ev.Name))
-	return false
+			},
+		},
+	)
 }
 
 // SyncExtraStages returns a map of history sync keys to [history.StageFunc] to be used for [history.Sync].
 //
 // Passing the return value of this method as the extraStages parameter to [history.Sync] results in forwarding events
 // from the Icinga DB history stream to Icinga Notifications after being resorted via the StreamSorter.
-func (client *Client) SyncExtraStages() map[string]history.StageFunc {
+func (client *Client) SyncExtraStages(ctx context.Context) map[string]history.StageFunc {
 	var syncKeyStructPtrs = map[string]any{
 		history.SyncPipelineAcknowledgement: (*v1history.AcknowledgementHistory)(nil),
 		history.SyncPipelineDowntime:        (*v1history.DowntimeHistoryMeta)(nil),
 		history.SyncPipelineFlapping:        (*v1history.FlappingHistory)(nil),
-		history.SyncPipelineState:           (*v1history.StateHistory)(nil),
 	}
 
 	sorterCallbackFn := func(msg redis.XMessage, key string) bool {
@@ -439,14 +493,13 @@ func (client *Client) SyncExtraStages() map[string]history.StageFunc {
 			return false
 		}
 
-		success := client.Submit(entity)
-		if success {
-			telemetry.Stats.NotificationSync.Add(1)
+		if err := client.Submit(ctx, entity); err == nil {
+			return true
 		}
-		return success
+		return false
 	}
 
-	pipelineFn := NewStreamSorter(client.ctx, client.logger, sorterCallbackFn).PipelineFunc
+	pipelineFn := NewStreamSorter(ctx, client.logger, sorterCallbackFn).PipelineFunc
 
 	extraStages := make(map[string]history.StageFunc)
 	for k := range syncKeyStructPtrs {