improve supervisor startup, add comments about issues with the supervisor getting stuck (#109)

* improve supervisor startup, add comments about issues with the supervisor getting stuck

## Improve supervisor startup

Previously, the supervisor's snapshot loop was not ideal for process startup:
  * We started a reflector and then *immediately* stopped it upon entering the snapshot loop.
  * Thus no updates would accumulate before creating the first snapshot. So the snapshot
    we uploaded would be basically a copy of the latest snapshot in S3.

Fix: We sleep first in the loop to allow updates to accumulate before stopping the reflector

Furthermore, the time-to-first-new-snapshot has actually been improved, perhaps
counter-intuitively given that we are now immediately *sleeping* instead of
making a snapshot.
  * Consider that previously the first fruitful iteration was delayed until the
    first (duplicate) snapshot was compressed & uploaded.

## Supervisor getting stuck

This PR also documents a "supervisor stuck" issue we hit:
  * We noticed the supervisor not making progress during a massive LDB cleanup.
  * We had a series of 31 ~1MB transactions that deleted an exorbitant number of stranded
    rows from the LDB.
  * To overcome the system being stuck we increased our configured sleep duration from
    5 to 30 minutes, until the system got unwedged.

* update tests to wait for first sleep to happen for snapshot to complete

* fix up language in comment

* already have a metric for ledger updates applied, just were not graphing it yet

Author: erikdw

pkg/supervisor/supervisor.go

@@ -129,22 +129,41 @@ func (s *supervisor) Start(ctx context.Context) {
 	events.Log("Starting supervisor")
 	s.reflectorCtl.Start(ctx)
 	defer events.Log("Stopped Supervisor")
+	sleepDur := s.SleepDuration
 	for {
-		sleepDur := s.SleepDuration
-		err := s.snapshot(ctx)
-		if err != nil && errors.Cause(err) != context.Canceled {
-			s.incrementSnapshotErrorMetric(1)
-			events.Log("Error taking snapshot: %{error}+v", err)
-			// Use a shorter sleep duration for faster retries
-			sleepDur = s.BreatheDuration
-		}
+		// Wait for the reflector to make changes to its LDB before stopping it.  Sometimes
+		// we need more time than to process a big transaction, such as a large cleanup of
+		// the LDB that does a deletion of many rows.
+		//
+		// ⚠️ If you don't increase the sleep duration enough to allow a transaction to complete,
+		// then the supervisor will get stuck and never make progress.
+		// Such a situation has a growing compound effect on the Ctlstore ecosystem,
+		// as all new reflectors have old snapshots and thus need to pull down an ever-growing
+		// number of ledger updates to sync-up with the latest state.
+		// This also creates an ever-increasing load on the Ctlstore DB (ctldb).
+		//
+		// TODO:
+		//   * Add detection of progress stopping (same ledger seq on 2+ runs) and backoff the sleep duration
+		//     automatically, up to a limit.
+		//       * Would be nice to be able to determine what the state of the LDB actions were
+		//         before we stopped the reflector and did the snapshot, so we can log it and spit out
+		//         a metric about it.
 		select {
 		case <-time.After(sleepDur):
+			// reset to default
+			sleepDur = s.SleepDuration
 		case <-ctx.Done():
 			events.Log("Supervisor exiting because context done (err=%v)", ctx.Err())
 			// Outer context is done, aborting everything
 			return
 		}
+		err := s.snapshot(ctx)
+		if err != nil && errors.Cause(err) != context.Canceled {
+			s.incrementSnapshotErrorMetric(1)
+			events.Log("Error taking snapshot: %{error}+v", err)
+			// Use a shorter sleep duration for faster retries
+			sleepDur = s.BreatheDuration
+		}
 	}
 }
 

pkg/supervisor/supervisor_test.go

@@ -50,8 +50,10 @@ func TestSupervisor(t *testing.T) {
 		require.True(t, reflector.Closed.IsSet())
 	}()
 
+	snapshotInterval := 100 * time.Millisecond
+
 	cfg := SupervisorConfig{
-		SnapshotInterval: 100 * time.Millisecond,
+		SnapshotInterval: snapshotInterval,
 		SnapshotURL:      "file://" + archivePath,
 		LDBPath:          ldbDbPath,
 		Reflector:        reflector,
@@ -127,7 +129,7 @@ func TestSupervisor(t *testing.T) {
 		}
 
 		// Wait for snapshot to complete
-		time.Sleep(10 * time.Millisecond)
+		time.Sleep(snapshotInterval)
 
 		// Cancels the context passed to the supervisor, which should cause it
 		// to return from the Start() call