exp/ingest/pipeline: Fix pipeline data race during shutdown

[bartekn]

PR Checklist

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, .md
  files, etc... affected by this change). Take a look in the docs folder for a given service,
  like this one.

Release planning

• I've updated the relevant CHANGELOG (here 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, or if it's mainly a patch change. The PR is targeted at the next
  release branch if it's not a patch change.

What

This commit fixes data race in exp/ingest/pipeline that can occur when LiveSession (and Horizon) is shut down.

It also removes updateStats method that was known to have a data race (see comment in that method). It is not actively used right now but was being reported by race detector.

Fix #2046.

Why

Previous code handling shutdown signal in LiveSession can be found below:

go/exp/ingest/live_session.go

Lines 196 to 217 in 5e4d247

	errChan := s.LedgerPipeline.Process(ledgerReader)
	select {
	case err2 := <-errChan:
	if err2 != nil {
	// Return with no errors if pipeline shutdown
	if err2 == pipeline.ErrShutdown {
	s.LedgerReporter.OnEndLedger(nil, true)
	return nil
	}
	if s.LedgerReporter != nil {
	s.LedgerReporter.OnEndLedger(err2, false)
	}
	return errors.Wrap(err2, "Ledger pipeline errored")
	}
	case <-s.standardSession.shutdown:
	if s.LedgerReporter != nil {
	s.LedgerReporter.OnEndLedger(nil, true)
	}
	s.LedgerPipeline.Shutdown()
	return nil
	}

The problem is when shutdown signal is received, Resume returns nil so Horizon starts it's shutdown code which calls Rollback() (using internal tx object) but at the same time pipeline is still running until the code receiving from ctx.Done channel is executed. It means that pipeline processors can execute transactions using tx transaction object in DB session. See #2046 for examples.

To fix this:

1. We don't select ingest session shutdown signal when waiting for pipeline to finish processing.
2. Instead we call Shutdown on pipelines inside LiveSession.Shutdown.
3. Then we wait/block until pipelines gracefully shutdown by calling Pipeline.IsRunning method.
4. Finally we close(s.shutdown) inside expingest/System.Shutdown().

So the components now shut down exactly in the following order:

1. Pipelines.
2. Session.
3. Horizon Expingest System.

One comment on -1 change in tests. When ingestSession.Run() returns nil we shouldn't continue to ingestSession.Resume() because nil value means that session ended. I updated the comment in LiveSession and also fixed Horizon code.

Known limitations

Pipeline design is very powerful but it's also very easy to introduce data races like this one. We may want to refactor this as noted previously in #2050.

[abuiles]

LGTM!

[abuiles]

@bartekn did we forget to add this before?

[bartekn]

Yes, there are two entry points to LiveSession: Run or Resume.