ingest/ledgerbackend: Fix captive core bug where --start-at-hash para…

…meter is omitted (#3265)

In #3201 we assumed it was possible to omit the --start-at-hash parameter when starting from the genesis ledger. However, Stellar Core actually fails when attempting that command. The --start-at-hash parameter is always required when using --start-at-ledger.

This means that if we want to stream from the genesis ledger we need to determine the either the horizon db or the history archives.

ingest/ledgerbackend/captive_core_backend.go

@@ -305,30 +305,28 @@ func (c *CaptiveStellarCore) runFromParams(from uint32) (runFrom uint32, ledgerH
 	}
 
 	if from <= 63 {
-		// For ledgers before (and including) first checkpoint, we start streaming
-		// without providing a hash, to avoid waiting for the checkpoint.
-		// It will always start streaming from ledger 2.
+		// For ledgers before (and including) first checkpoint, get/wait the first
+		// checkpoint to get the ledger header. It will always start streaming
+		// from ledger 2.
 		nextLedger = 2
-		runFrom = 2
 		// The line below is to support a special case for streaming ledger 2
 		// that works for all other ledgers <= 63 (fast-forward).
 		// We can't set from=2 because Stellar-Core will not allow starting from 1.
 		// To solve this we start from 3 and exploit the fast that Stellar-Core
 		// will stream data from 2 for the first checkpoint.
 		from = 3
-		return
-	}
-
-	// For ledgers after the first checkpoint, start at the previous checkpoint
-	// and fast-forward from there.
-	if !historyarchive.IsCheckpoint(from) {
-		from = historyarchive.PrevCheckpoint(from)
-	}
-	// Streaming will start from the previous checkpoint + 1
-	nextLedger = from - 63
-	if nextLedger < 2 {
-		// Stellar-Core always streams from ledger 2 at min.
-		nextLedger = 2
+	} else {
+		// For ledgers after the first checkpoint, start at the previous checkpoint
+		// and fast-forward from there.
+		if !historyarchive.IsCheckpoint(from) {
+			from = historyarchive.PrevCheckpoint(from)
+		}
+		// Streaming will start from the previous checkpoint + 1
+		nextLedger = from - 63
+		if nextLedger < 2 {
+			// Stellar-Core always streams from ledger 2 at min.
+			nextLedger = 2
+		}
 	}
 
 	runFrom = from - 1

ingest/ledgerbackend/captive_core_backend_test.go

@@ -978,6 +978,8 @@ func TestCaptiveUseOfLedgerHashStore(t *testing.T) {
 		Return("cde", true, nil).Once()
 	mockLedgerHashStore.On("GetLedgerHash", uint32(126)).
 		Return("ghi", true, nil).Once()
+	mockLedgerHashStore.On("GetLedgerHash", uint32(2)).
+		Return("mnb", true, nil).Once()
 
 	captiveBackend := CaptiveStellarCore{
 		archive:           mockArchive,
@@ -988,7 +990,7 @@ func TestCaptiveUseOfLedgerHashStore(t *testing.T) {
 	runFrom, ledgerHash, nextLedger, err := captiveBackend.runFromParams(24)
 	assert.NoError(t, err)
 	assert.Equal(t, uint32(2), runFrom)
-	assert.Equal(t, "", ledgerHash)
+	assert.Equal(t, "mnb", ledgerHash)
 	assert.Equal(t, uint32(2), nextLedger)
 
 	runFrom, ledgerHash, nextLedger, err = captiveBackend.runFromParams(86)
@@ -1064,11 +1066,7 @@ func TestCaptiveRunFromParams(t *testing.T) {
 			runFrom, ledgerHash, nextLedger, err := captiveBackend.runFromParams(tc.from)
 			tt.NoError(err)
 			tt.Equal(tc.runFrom, runFrom, "runFrom")
-			if tc.from <= 63 {
-				tt.Empty(ledgerHash)
-			} else {
-				tt.Equal("0101010100000000000000000000000000000000000000000000000000000000", ledgerHash)
-			}
+			tt.Equal("0101010100000000000000000000000000000000000000000000000000000000", ledgerHash)
 			tt.Equal(tc.nextLedger, nextLedger, "nextLedger")
 		})
 	}

ingest/ledgerbackend/stellar_core_runner.go

@@ -263,17 +263,13 @@ func (r *stellarCoreRunner) runFrom(from uint32, hash string) error {
 		return errors.New("runner already started")
 	}
 	var err error
-	args := []string{
+	r.cmd, err = r.createCmd(
 		"run",
 		"--in-memory",
 		"--start-at-ledger", fmt.Sprintf("%d", from),
+		"--start-at-hash", hash,
 		"--metadata-output-stream", r.getPipeName(),
-	}
-	if hash != "" {
-		args = append(args, "--start-at-hash", hash)
-	}
-
-	r.cmd, err = r.createCmd(args...)
+	)
 	if err != nil {
 		return errors.Wrap(err, "error creating `stellar-core run` subprocess")
 	}

services/horizon/internal/test/integration/integration.go

@@ -215,9 +215,8 @@ func (i *Test) waitForHorizon() {
 			continue
 		}
 
-		if root.HorizonSequence < 2 ||
-			int(root.HorizonSequence) != int(root.IngestSequence) ||
-			root.HorizonSequence < root.CoreSequence {
+		if root.HorizonSequence < 3 ||
+			int(root.HorizonSequence) != int(root.IngestSequence) {
 			i.t.Logf("Horizon ingesting... %v", root)
 			time.Sleep(time.Second)
 			continue