Introduce instance ready hook

go.mod

@@ -1,6 +1,6 @@
 module github.com/PowerDNS/lightningstream
 
-go 1.24.0
+go 1.24.13
 
 require (
 	github.com/CrowdStrike/csproto v0.35.0

syncer/hooks/hooks.go

@@ -40,6 +40,11 @@ type Hooks struct {
 
 	// Returns a channel that can inject other updates into the sync loop
 	OtherUpdateSource func() <-chan snapshot.Update
+
+	// InstanceReady is called to check whether, after application of this
+	// update, an instance can be considered to be "ready". This will signal the
+	// health check to become ready or the OnlyOnce sync to be done.
+	InstanceReady func(*snapshot.NameInfo) bool
 }
 
 type SnapshotInfo struct {

syncer/receiver/downloader.go

@@ -116,7 +116,7 @@ func (d *Downloader) LoadOnce(ctx context.Context, ni snapshot.NameInfo) error {
 
 	msg, err := snapshot.LoadData(data)
 	if err != nil {
-		d.l.Debug("Returning DecompressedSnapshotToken")
+		d.l.Debug("Releasing DecompressedSnapshotToken")
 		token.Release()
 		// This snapshot is considered corrupt, we will ignore it from now on
 		d.r.MarkCorrupt(ni.FullName, err)
@@ -144,7 +144,7 @@ func (d *Downloader) LoadOnce(ctx context.Context, ni snapshot.NameInfo) error {
 			if u.Snapshot == nil {
 				return // already called?
 			}
-			d.l.Debug("Returning DecompressedSnapshotToken")
+			d.l.Debug("Releasing DecompressedSnapshotToken")
 			// Clear it before returning the token
 			u.Snapshot = nil
 			utils.GC()

syncer/sync.go

@@ -199,23 +199,36 @@ func (s *Syncer) syncLoop(ctx context.Context, env *lmdb.Env, r *receiver.Receiv
 			}
 
 			// New update to load
+			l := s.l.WithFields(logrus.Fields{
+				"kind": update.NameInfo.Kind,
+				"file": update.NameInfo.FullName,
+			})
 			if instance == ownInstanceID {
-				s.l.WithField("kind", update.NameInfo.Kind).Info("Loading update for own instance")
+				l.Info("Loading update for own instance")
+			} else {
+				l.Debug("Loading update")
 			}
+			l = s.l.WithField("other_instance", instance)
+
 			if update.NameInfo.Kind == snapshot.KindSnapshot {
 				nLoads++
-				if waitingForInstances.Contains(instance) {
-					s.l.WithField("other_instance", instance).Info("No longer waiting for instance")
+				if waitingForInstances.Contains(instance) && s.hooks.InstanceReady == nil {
+					l.Info("No longer waiting for instance")
 					waitingForInstances.Remove(instance)
 				}
 			}
-			s.l.WithFields(logrus.Fields{
-				"kind": update.NameInfo.Kind,
-				"file": update.NameInfo.FullName,
-			}).Debug("Loading update")
+			if waitingForInstances.Contains(instance) && s.hooks.InstanceReady != nil {
+				if s.hooks.InstanceReady(&update.NameInfo) {
+					l.Info("No longer waiting for instance")
+					waitingForInstances.Remove(instance)
+				} else {
+					l.Info("Waiting for additional updates for instance")
+				}
+			}
+
 			actualTxnID, localChanged, err := s.LoadOnce(
 				ctx, env, instance, update, lastSyncedTxnID)
-			update.Close() // returns the DecompressedSnapshotToken
+			update.Close() // releases the DecompressedSnapshotToken
 			if err != nil {
 				return err
 			}
@@ -225,9 +238,6 @@ func (s *Syncer) syncLoop(ctx context.Context, env *lmdb.Env, r *receiver.Receiv
 				NameInfo: update.NameInfo,
 			})
 
-			if update.NameInfo.Kind == snapshot.KindSnapshot {
-				utils.GC()
-			}
 			if !localChanged {
 				// Prevent triggering a local snapshot if there were no local
 				// changes by bumping the transaction ID we consider synced
@@ -329,16 +339,13 @@ func (s *Syncer) syncLoop(ctx context.Context, env *lmdb.Env, r *receiver.Receiv
 		}
 
 		// Sleep before next check for snapshots and local changes
-		//s.l.Debug("Waiting for a new transaction")
 		if err := utils.SleepContext(ctx, s.c.LMDBPollInterval); err != nil {
 			return err
 		}
 	}
-
 }
 
 func (s *Syncer) LoadOnce(ctx context.Context, env *lmdb.Env, instance string, update snapshot.Update, lastTxnID header.TxnID) (txnID header.TxnID, localChanged bool, err error) {
-
 	t0 := time.Now() // for performance measurements
 	snap := update.Snapshot
 
@@ -427,7 +434,7 @@ func (s *Syncer) LoadOnce(ctx context.Context, env *lmdb.Env, instance string, u
 							dbiName, snap.FormatVersion)
 					}
 
-					var flags = dbiflags.Flags(dbiMsg.Flags())
+					flags := dbiflags.Flags(dbiMsg.Flags())
 					if dbiOpt.OverrideCreateFlags != nil {
 						flags = *dbiOpt.OverrideCreateFlags
 					}
@@ -448,7 +455,7 @@ func (s *Syncer) LoadOnce(ctx context.Context, env *lmdb.Env, instance string, u
 				// The formatVersion does not matter here, because the DBI flags
 				// stored in earlier versions will be the correct ones for the
 				// DBI that we are creating here (shadow or native).
-				var flags = dbiflags.Flags(dbiMsg.Flags())
+				flags := dbiflags.Flags(dbiMsg.Flags())
 				if dbiOpt.OverrideCreateFlags != nil {
 					flags = *dbiOpt.OverrideCreateFlags
 				}