Repository: stellar/go
Pull request #5307 (2024-05-09 10:15:19)

ingest/ledgerbackend: Improve thread-safety of stellarCoreRunner.close()
<!-- If you're making a doc PR or something tiny where the below is irrelevant, delete this
template and use a short description, but in your description aim to include both what the
change is, and why it is being made, with enough context for anyone to understand. -->

<details>
  <summary>PR Checklist</summary>
  
### PR Structure

* [ ] This PR has reasonably narrow scope (if not, break it down into smaller PRs).
* [ ] This PR avoids mixing refactoring changes with feature changes (split into two PRs
  otherwise).
* [ ] This PR's title starts with name of package that is most changed in the PR, ex.
  `services/friendbot`, or `all` or `doc` if the changes are broad or impact many
  packages.

### Thoroughness

* [ ] This PR adds tests for the most critical parts of the new functionality or fixes.
* [ ] I've updated any docs ([developer docs](https://developers.stellar.org/api/), `.md`
  files, etc... affected by this change). Take a look in the `docs` folder for a given service,
  like [this one](https://github.com/stellar/go/tree/master/services/horizon/internal/docs).

### Release planning

* [ ] I've updated the relevant CHANGELOG ([here](services/horizon/CHANGELOG.md) for Horizon) if
  needed with deprecations, added features, breaking changes, and DB schema changes.
* [ ] I've decided if this PR requires a new major/minor version according to
  [semver](https://semver.org/), or if it's mainly a patch change. The PR is targeted at the next
  release branch if it's not a patch change.
</details>

### What

While looking at ingestion logs I observed an unexpected error `error preparing range: error starting prepare range: the previous Stellar-Core instance is still running` which occurred when we rolled out a new version of core:

```
 level=warning msg="detected new version of captive core binary /usr/bin/stellar-core , aborting session." service=ingest subservice=stellar-core
 level=info msg="default: got signal 2, shutting down" service=ingest subservice=stellar-core
 level=info msg="Processed ledger" commit=true duration=0.732382788 ledger=true sequence=51570776 service=ingest state=true
 level=info msg="Ingestion system state machine transition" current_state="resume(latestSuccessfullyProcessedLedger=51570775)" next_state="resume(latestSuccessfullyProcessedLedger=51570776)" service=ingest level=error msg="Error in ingestion state machine" current_state="resume(latestSuccessfullyProcessedLedger=51570776)" error="error preparing range: error starting prepare range: the previous Stellar-Core instance is still running" next_state=start service=ingest level=info msg="default: Application destructing" service=ingest subservice=stellar-core
 level=info msg="default: Application destroyed" pid=31974 service=ingest subservice=stellar-core
```

The error occurred in this block of code:

https://github.com/stellar/go/blob/ab3a9265a31d7a86a794d447061652ab53cc3237/ingest/ledgerbackend/captive_core_backend.go#L456-L460

I was surprised that stellar-core was still running even though we just called `c.stellarCoreRunner.close()`. I expected the `close()` function on stellarCoreRunner to block until the core process was reaped. However, it turns out that is not the case in this scenario because the `close()` function is called by multiple go routines.

Once the new core binary is detected, the file watcher go routine will first call `close()` https://github.com/stellar/go/blob/ab3a9265a31d7a86a794d447061652ab53cc3237/ingest/ledgerbackend/file_watcher.go#L66 . This invocation will block until the core process is reaped. However, it takes a few seconds for that to happen. So, while that is occurring PrepareRange() calls `close()` and that returns immediately based on this condition:

https://github.com/stellar/go/blob/ab3a9265a31d7a86a794d447061652ab53cc3237/ingest/ledgerbackend/stellar_core_runner.go#L555-L559

This PR fixes this issue by guaranteeing that the core process will be reaped after any invocation of `close()` , even if `close()` is called concurrently. This behavior is implemented by using https://pkg.go.dev/sync#Once

### Why

This bug does not prevent ingestion from progressing because upon errors we will still retry to ingest and eventually `PrepareRange()` is able to succeed once the core process is reaped. However, it is still good to get rid of this error case for the following reasons:

- reduce unnecessary delays caused by ingestion retries
- reduce error messages in the ingestion logs
- we will soon alert on ingestion errors so it's good to remove cases which should not actually be errors

### Known limitations

[N/A]



Patch:
diff --git a/ingest/ledgerbackend/captive_core_backend.go b/ingest/ledgerbackend/captive_core_backend.go
index 50e933bb..25b22d01 100644
--- a/ingest/ledgerbackend/captive_core_backend.go
+++ b/ingest/ledgerbackend/captive_core_backend.go
@@ -452,12 +452,6 @@ func (c *CaptiveStellarCore) startPreparingRange(ctx context.Context, ledgerRang
 		if err := c.stellarCoreRunner.close(); err != nil {
 			return false, errors.Wrap(err, "error closing existing session")
 		}
-
-		// Make sure Stellar-Core is terminated before starting a new instance.
-		processExited, _ := c.stellarCoreRunner.getProcessExitError()
-		if !processExited {
-			return false, errors.New("the previous Stellar-Core instance is still running")
-		}
 	}
 
 	var err error
diff --git a/ingest/ledgerbackend/stellar_core_runner.go b/ingest/ledgerbackend/stellar_core_runner.go
index 1c2c09c4..7f883b69 100644
--- a/ingest/ledgerbackend/stellar_core_runner.go
+++ b/ingest/ledgerbackend/stellar_core_runner.go
@@ -17,6 +17,7 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
+
 	"github.com/stellar/go/protocols/stellarcore"
 	"github.com/stellar/go/support/log"
 )
@@ -65,6 +66,7 @@ type stellarCoreRunner struct {
 	systemCaller systemCaller
 
 	lock             sync.Mutex
+	closeOnce        sync.Once
 	processExited    bool
 	processExitError error
 
@@ -285,9 +287,6 @@ func (r *stellarCoreRunner) catchup(from, to uint32) error {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
-	r.mode = stellarCoreRunnerModeOffline
-	r.storagePath = r.getFullStoragePath()
-
 	// check if we have already been closed
 	if r.ctx.Err() != nil {
 		return r.ctx.Err()
@@ -297,6 +296,9 @@ func (r *stellarCoreRunner) catchup(from, to uint32) error {
 		return errors.New("runner already started")
 	}
 
+	r.mode = stellarCoreRunnerModeOffline
+	r.storagePath = r.getFullStoragePath()
+
 	rangeArg := fmt.Sprintf("%d/%d", to, to-from+1)
 	params := []string{"catchup", rangeArg, "--metadata-output-stream", r.getPipeName()}
 
@@ -350,9 +352,6 @@ func (r *stellarCoreRunner) runFrom(from uint32, hash string) error {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
-	r.mode = stellarCoreRunnerModeOnline
-	r.storagePath = r.getFullStoragePath()
-
 	// check if we have already been closed
 	if r.ctx.Err() != nil {
 		return r.ctx.Err()
@@ -362,6 +361,9 @@ func (r *stellarCoreRunner) runFrom(from uint32, hash string) error {
 		return errors.New("runner already started")
 	}
 
+	r.mode = stellarCoreRunnerModeOnline
+	r.storagePath = r.getFullStoragePath()
+
 	var err error
 
 	if r.useDB {
@@ -546,53 +548,45 @@ func (r *stellarCoreRunner) getProcessExitError() (bool, error) {
 // the necessary cleanup on the resources associated with the captive core process
 // close is both thread safe and idempotent
 func (r *stellarCoreRunner) close() error {
-	r.lock.Lock()
-	started := r.started
-	storagePath := r.storagePath
-
-	r.storagePath = ""
-
-	// check if we have already closed
-	if storagePath == "" {
+	var closeError error
+	r.closeOnce.Do(func() {
+		r.lock.Lock()
+		// we cancel the context while holding the lock in order to guarantee that
+		// this captive core instance cannot start once the lock is released.
+		// catchup() and runFrom() can only execute while holding the lock and if
+		// the context is canceled both catchup() and runFrom() will abort early
+		// without performing any side effects (e.g. state mutations).
+		r.cancel()
 		r.lock.Unlock()
-		return nil
-	}
 
-	if !started {
-		// Update processExited if handleExit that updates it not even started
-		// (error before command run).
-		r.processExited = true
-	}
-
-	r.cancel()
-	r.lock.Unlock()
+		// only reap captive core sub process and related go routines if we've started
+		// otherwise, just cleanup the temp dir
+		if r.started {
+			// wait for the stellar core process to terminate
+			r.wg.Wait()
 
-	// only reap captive core sub process and related go routines if we've started
-	// otherwise, just cleanup the temp dir
-	if started {
-		// wait for the stellar core process to terminate
-		r.wg.Wait()
+			// drain meta pipe channel to make sure the ledger buffer goroutine exits
+			for range r.getMetaPipe() {
 
-		// drain meta pipe channel to make sure the ledger buffer goroutine exits
-		for range r.getMetaPipe() {
+			}
 
+			// now it's safe to close the pipe reader
+			// because the ledger buffer is no longer reading from it
+			r.pipe.Reader.Close()
 		}
 
-		// now it's safe to close the pipe reader
-		// because the ledger buffer is no longer reading from it
-		r.pipe.Reader.Close()
-	}
-
-	if r.mode != 0 && (runtime.GOOS == "windows" ||
-		(r.processExitError != nil && r.processExitError != context.Canceled) ||
-		r.mode == stellarCoreRunnerModeOffline) {
-		// It's impossible to send SIGINT on Windows so buckets can become
-		// corrupted. If we can't reuse it, then remove it.
-		// We also remove the storage path if there was an error terminating the
-		// process (files can be corrupted).
-		// We remove all files when reingesting to save disk space.
-		return r.systemCaller.removeAll(storagePath)
-	}
+		if r.mode != 0 && (runtime.GOOS == "windows" ||
+			(r.processExitError != nil && r.processExitError != context.Canceled) ||
+			r.mode == stellarCoreRunnerModeOffline) {
+			// It's impossible to send SIGINT on Windows so buckets can become
+			// corrupted. If we can't reuse it, then remove it.
+			// We also remove the storage path if there was an error terminating the
+			// process (files can be corrupted).
+			// We remove all files when reingesting to save disk space.
+			closeError = r.systemCaller.removeAll(r.storagePath)
+			return
+		}
+	})
 
-	return nil
+	return closeError
 }


Test patch:
diff --git a/ingest/ledgerbackend/captive_core_backend_test.go b/ingest/ledgerbackend/captive_core_backend_test.go
index 5178fd97..a367f560 100644
--- a/ingest/ledgerbackend/captive_core_backend_test.go
+++ b/ingest/ledgerbackend/captive_core_backend_test.go
@@ -302,8 +302,6 @@ func TestCaptivePrepareRangeCloseNotFullyTerminated(t *testing.T) {
 	mockRunner.On("getMetaPipe").Return((<-chan metaResult)(metaChan))
 	mockRunner.On("context").Return(ctx)
 	mockRunner.On("close").Return(nil)
-	mockRunner.On("getProcessExitError").Return(true, nil)
-	mockRunner.On("getProcessExitError").Return(false, nil)
 
 	mockArchive := &historyarchive.MockArchive{}
 	mockArchive.
diff --git a/ingest/ledgerbackend/mock_cmd_test.go b/ingest/ledgerbackend/mock_cmd_test.go
index a1d28042..a28b6c8d 100644
--- a/ingest/ledgerbackend/mock_cmd_test.go
+++ b/ingest/ledgerbackend/mock_cmd_test.go
@@ -70,7 +70,7 @@ func simpleCommandMock() *mockCmd {
 	cmdMock.On("getStdout").Return(writer)
 	cmdMock.On("setStderr", mock.Anything)
 	cmdMock.On("getStderr").Return(writer)
-	cmdMock.On("getProcess").Return(&os.Process{})
+	cmdMock.On("getProcess").Return(&os.Process{}).Maybe()
 	cmdMock.On("setExtraFiles", mock.Anything)
 	cmdMock.On("Start").Return(nil)
 	return cmdMock
diff --git a/ingest/ledgerbackend/stellar_core_runner_test.go b/ingest/ledgerbackend/stellar_core_runner_test.go
index 60871922..00cb2913 100644
--- a/ingest/ledgerbackend/stellar_core_runner_test.go
+++ b/ingest/ledgerbackend/stellar_core_runner_test.go
@@ -3,6 +3,7 @@ package ledgerbackend
 import (
 	"context"
 	"encoding/json"
+	"sync"
 	"testing"
 	"time"
 
@@ -46,7 +47,7 @@ func TestCloseOffline(t *testing.T) {
 		"fd:3",
 		"--in-memory",
 	).Return(cmdMock)
-	scMock.On("removeAll", mock.Anything).Return(nil)
+	scMock.On("removeAll", mock.Anything).Return(nil).Once()
 	runner.systemCaller = scMock
 
 	assert.NoError(t, runner.catchup(100, 200))
@@ -133,7 +134,7 @@ func TestCloseOnlineWithError(t *testing.T) {
 		"--metadata-output-stream",
 		"fd:3",
 	).Return(cmdMock)
-	scMock.On("removeAll", mock.Anything).Return(nil)
+	scMock.On("removeAll", mock.Anything).Return(nil).Once()
 	runner.systemCaller = scMock
 
 	assert.NoError(t, runner.runFrom(100, "hash"))
@@ -149,6 +150,61 @@ func TestCloseOnlineWithError(t *testing.T) {
 	assert.NoError(t, runner.close())
 }
 
+func TestCloseConcurrency(t *testing.T) {
+	captiveCoreToml, err := NewCaptiveCoreToml(CaptiveCoreTomlParams{})
+	assert.NoError(t, err)
+
+	captiveCoreToml.AddExamplePubnetValidators()
+
+	runner := newStellarCoreRunner(CaptiveCoreConfig{
+		BinaryPath:         "/usr/bin/stellar-core",
+		HistoryArchiveURLs: []string{"http://localhost"},
+		Log:                log.New(),
+		Context:            context.Background(),
+		Toml:               captiveCoreToml,
+		StoragePath:        "/tmp/captive-core",
+	})
+
+	cmdMock := simpleCommandMock()
+	cmdMock.On("Wait").Return(errors.New("wait error")).WaitUntil(time.After(time.Millisecond * 300))
+	defer cmdMock.AssertExpectations(t)
+
+	// Replace system calls with a mock
+	scMock := &mockSystemCaller{}
+	defer scMock.AssertExpectations(t)
+	scMock.On("stat", mock.Anything).Return(isDirImpl(true), nil)
+	scMock.On("writeFile", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	scMock.On("command",
+		"/usr/bin/stellar-core",
+		"--conf",
+		mock.Anything,
+		"--console",
+		"catchup",
+		"200/101",
+		"--metadata-output-stream",
+		"fd:3",
+		"--in-memory",
+	).Return(cmdMock)
+	scMock.On("removeAll", mock.Anything).Return(nil).Once()
+	runner.systemCaller = scMock
+
+	assert.NoError(t, runner.catchup(100, 200))
+
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			assert.NoError(t, runner.close())
+			exited, err := runner.getProcessExitError()
+			assert.True(t, exited)
+			assert.Error(t, err)
+		}()
+	}
+
+	wg.Wait()
+}
+
 func TestRunFromUseDBLedgersMatch(t *testing.T) {
 	captiveCoreToml, err := NewCaptiveCoreToml(CaptiveCoreTomlParams{})
 	assert.NoError(t, err)
@@ -300,7 +356,7 @@ func TestRunFromUseDBLedgersInFront(t *testing.T) {
 	scMock := &mockSystemCaller{}
 	defer scMock.AssertExpectations(t)
 	// Storage dir is removed because ledgers do not match
-	scMock.On("removeAll", mock.Anything).Return(nil)
+	scMock.On("removeAll", mock.Anything).Return(nil).Once()
 	scMock.On("stat", mock.Anything).Return(isDirImpl(true), nil)
 	scMock.On("writeFile", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	scMock.On("command",