ingest/ledgerbackend: Handle user initiated shutdown during catchup #3258
+442
−213
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,7 +1,9 @@ | ||
| package ledgerbackend | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/hex" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/pkg/errors" | ||
| @@ -75,9 +77,16 @@ type CaptiveStellarCore struct { | ||
| ledgerBuffer *bufferedLedgerMetaReader | ||
| stellarCoreRunner stellarCoreRunnerInterface | ||
|
|
||
| // waitIntervalPrepareRange defines a time to wait between checking if the buffer | ||
| // is empty. Default 1s, lower in tests to make them faster. | ||
| waitIntervalPrepareRange time.Duration | ||
|
|
||
| // For testing | ||
| stellarCoreRunnerFactory func(mode stellarCoreRunnerMode) (stellarCoreRunnerInterface, error) | ||
|
|
||
| // mutex protects all values below, ex. Close() can be called concurrently. | ||
| mutex sync.Mutex | ||
|
|
||
| // Defines if the blocking mode (off by default) is on or off. In blocking mode, | ||
| // calling GetLedger blocks until the requested ledger is available. This is useful | ||
| // for scenarios when Horizon consumes ledgers faster than Stellar-Core produces them | ||
| @@ -91,10 +100,6 @@ type CaptiveStellarCore struct { | ||
| nextLedger uint32 // next ledger expected, error w/ restart if not seen | ||
| lastLedger *uint32 // end of current segment if offline, nil if online | ||
| previousLedgerHash *string | ||
| } | ||
|
|
||
| // CaptiveCoreConfig contains all the parameters required to create a CaptiveStellarCore instance | ||
| @@ -134,7 +139,7 @@ func NewCaptive(config CaptiveCoreConfig) (*CaptiveStellarCore, error) { | ||
| c := &CaptiveStellarCore{ | ||
| archive: archive, | ||
| configAppendPath: config.ConfigAppendPath, | ||
| waitIntervalPrepareRange: 100 * time.Millisecond, | ||
| ledgerHashStore: config.LedgerHashStore, | ||
| } | ||
| c.stellarCoreRunnerFactory = func(mode stellarCoreRunnerMode) (stellarCoreRunnerInterface, error) { | ||
| @@ -153,7 +158,7 @@ func (c *CaptiveStellarCore) getLatestCheckpointSequence() (uint32, error) { | ||
| } | ||
|
|
||
| func (c *CaptiveStellarCore) openOfflineReplaySubprocess(from, to uint32) error { | ||
| err := c.close() | ||
| if err != nil { | ||
| return errors.Wrap(err, "error closing existing session") | ||
| } | ||
| @@ -202,7 +207,7 @@ func (c *CaptiveStellarCore) openOnlineReplaySubprocess(from uint32) error { | ||
| return nil | ||
| } | ||
|
|
||
| err := c.close() | ||
| if err != nil { | ||
| return errors.Wrap(err, "error closing existing session") | ||
| } | ||
| @@ -258,18 +263,6 @@ func (c *CaptiveStellarCore) openOnlineReplaySubprocess(from uint32) error { | ||
| // read-ahead buffer | ||
| c.ledgerBuffer = newBufferedLedgerMetaReader(c.stellarCoreRunner) | ||
| go c.ledgerBuffer.start(0) | ||
| return nil | ||
| } | ||
|
|
||
| @@ -340,11 +333,19 @@ func (c *CaptiveStellarCore) runFromParams(from uint32) (runFrom uint32, ledgerH | ||
| // it normally (including connecting to the Stellar network). | ||
| // Please note that using a BoundedRange, currently, requires a full-trust on | ||
| // history archive. This issue is being fixed in Stellar-Core. | ||
| // | ||
| // Returns `context.Cancelled` if `Close` was called from another Go routine. | ||
| func (c *CaptiveStellarCore) PrepareRange(ledgerRange Range) error { | ||
| // c.mutex is unlocked on error and before waitloop to allow calling Close() | ||
| // to intrerupt the process. | ||
| c.mutex.Lock() | ||
|
|
||
| // Range already prepared | ||
| if prepared, err := c.isPrepared(ledgerRange); err != nil { | ||
| c.mutex.Unlock() | ||
|
There was a problem hiding this comment. Would it be safer / cleaner to do something like defer func() {
if !unlocked {
c.mutex.Unlock()
unlocked = true
}
}instead (and set It just seems like it could be really easy for someone (like me lol) to edit this code in the future, then forget to unlock the mutex and be in a dangerous state. |
||
| return errors.Wrap(err, "error in IsPrepared") | ||
| } else if prepared { | ||
| c.mutex.Unlock() | ||
| return nil | ||
| } | ||
|
|
||
| @@ -355,24 +356,51 @@ func (c *CaptiveStellarCore) PrepareRange(ledgerRange Range) error { | ||
| err = c.openOnlineReplaySubprocess(ledgerRange.from) | ||
| } | ||
| if err != nil { | ||
| c.mutex.Unlock() | ||
| return errors.Wrap(err, "opening subprocess") | ||
| } | ||
|
|
||
| c.mutex.Unlock() | ||
|
|
||
| waitloop: | ||
| for { | ||
| select { | ||
| case <-c.stellarCoreRunner.getTomb().Dead(): | ||
bartekn marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| // If core is dead we need to handle the Err() value: | ||
| // - nil and unbounded range: return error because we don't expect | ||
| // Stellar-Core to exit while streaming unbounded range, | ||
| // - nil and bounded range: this is expeced in bounded range because | ||
| // Stellar-Core can exit after streaming all ledgers, break and | ||
| // wait for buffer to fill, | ||
| // - context.Cancelled: user initiated exit, return the same error, | ||
| // - not nil: exit with error, wrap and return. | ||
| processErr := c.stellarCoreRunner.getTomb().Err() | ||
| switch { | ||
| case processErr == nil && !ledgerRange.bounded: | ||
| return errors.New("stellar-core process exited unexpectedly without an error") | ||
| case processErr == nil && ledgerRange.bounded: | ||
| return nil | ||
| case processErr == context.Canceled: | ||
| return processErr | ||
| default: | ||
| return errors.Wrap(processErr, "stellar-core process exited with an error") | ||
| } | ||
|
Comment on lines
+377
to
387
There was a problem hiding this comment.
|
||
| default: | ||
| } | ||
| // Wait/fast-forward to the expected ledger or an error. We need to check | ||
| // buffer length because `GetLedger` may be blocking. | ||
|
There was a problem hiding this comment. I think it would make sense to add a context to There was a problem hiding this comment. I really wanted to avoid using
There was a problem hiding this comment. Sure. I will open a new issue. |
||
| for len(c.ledgerBuffer.getChannel()) > 0 { | ||
| _, _, err := c.GetLedger(c.nextLedger) | ||
| if err != nil { | ||
| return errors.Wrapf(err, "Error fast-forwarding to %d", ledgerRange.from) | ||
| } | ||
|
|
||
| // If nextLedger > ledgerRange.from then ledgerRange.from is cached. | ||
| if c.nextLedger > ledgerRange.from { | ||
| break waitloop | ||
| } | ||
| } | ||
|
|
||
| time.Sleep(c.waitIntervalPrepareRange) | ||
|
There was a problem hiding this comment. Do we really need an explicit sleep here? I would (at the very least) use a ticker and intertwine the check with the other channels in the Even better, It may even be possible to remove the wait and condition by using an asynchronous go-routine for fast-forwarding. It may seem outside of the scope of this PR, but it affects the shutdown wait time. There was a problem hiding this comment. We can try it out for sure! Can you create a new issue? I'm not sure if we'll be able to test it before beta. |
||
| } | ||
|
|
||
| @@ -381,6 +409,12 @@ func (c *CaptiveStellarCore) PrepareRange(ledgerRange Range) error { | ||
|
|
||
| // IsPrepared returns true if a given ledgerRange is prepared. | ||
| func (c *CaptiveStellarCore) IsPrepared(ledgerRange Range) (bool, error) { | ||
| c.mutex.Lock() | ||
| defer c.mutex.Unlock() | ||
| return c.isPrepared(ledgerRange) | ||
| } | ||
|
|
||
| func (c *CaptiveStellarCore) isPrepared(ledgerRange Range) (bool, error) { | ||
| lastLedger := uint32(0) | ||
| if c.lastLedger != nil { | ||
| lastLedger = *c.lastLedger | ||
| @@ -391,10 +425,10 @@ func (c *CaptiveStellarCore) IsPrepared(ledgerRange Range) (bool, error) { | ||
| cachedLedger = c.cachedMeta.LedgerSequence() | ||
| } | ||
|
|
||
| return c.isPreparedParams(c.nextLedger, lastLedger, cachedLedger, ledgerRange), nil | ||
| } | ||
|
|
||
| func (*CaptiveStellarCore) isPreparedParams(nextLedger, lastLedger, cachedLedger uint32, ledgerRange Range) bool { | ||
| if nextLedger == 0 { | ||
| return false | ||
| } | ||
| @@ -431,7 +465,15 @@ func (*CaptiveStellarCore) isPrepared(nextLedger, lastLedger, cachedLedger uint3 | ||
| // * UnboundedRange makes GetLedger non-blocking. The method will return with | ||
| // the first argument equal false. | ||
| // This is done to provide maximum performance when streaming old ledgers. | ||
| // | ||
| // Returns `context.Cancelled` if `Close` was called from another Go routine. | ||
| func (c *CaptiveStellarCore) GetLedger(sequence uint32) (bool, xdr.LedgerCloseMeta, error) { | ||
| c.mutex.Lock() | ||
| defer c.mutex.Unlock() | ||
| return c.getLedger(sequence) | ||
| } | ||
|
|
||
| func (c *CaptiveStellarCore) getLedger(sequence uint32) (bool, xdr.LedgerCloseMeta, error) { | ||
| if c.cachedMeta != nil && sequence == c.cachedMeta.LedgerSequence() { | ||
| // GetLedger can be called multiple times using the same sequence, ex. to create | ||
| // change and transaction readers. If we have this ledger buffered, let's return it. | ||
| @@ -466,7 +508,7 @@ loop: | ||
| return false, xdr.LedgerCloseMeta{}, nil | ||
|
There was a problem hiding this comment. If Close() is called concurrently with this function is there a possible race condition where the ledger buffer is set to nil right before it is accessed here? There was a problem hiding this comment. Good catch, I focused on shutdown code and forgot about the obvious things. Fixed in 648ebf5. |
||
| } | ||
|
|
||
| // We don't have to handle Stellar-Core exit because this is handled | ||
| // in bufferedLedgerMetaReader (will send an error to the channel). | ||
| result := <-c.ledgerBuffer.getChannel() | ||
| if result.err != nil { | ||
| @@ -507,7 +549,7 @@ loop: | ||
|
|
||
| // If we got the _last_ ledger in a segment, close before returning. | ||
| if c.lastLedger != nil && *c.lastLedger == seq { | ||
| if err := c.close(); err != nil { | ||
| return false, xdr.LedgerCloseMeta{}, errors.Wrap(err, "error closing session") | ||
| } | ||
| } | ||
| @@ -517,7 +559,7 @@ loop: | ||
| // All paths above that break out of the loop (instead of return) | ||
| // set e to non-nil: there was an error and we should close and | ||
| // reset state before retuning an error to our caller. | ||
| c.close() | ||
| return false, xdr.LedgerCloseMeta{}, errOut | ||
| } | ||
|
|
||
| @@ -529,6 +571,9 @@ loop: | ||
| // the latest sequence closed by the network. It's always the last value available | ||
| // in the backend. | ||
| func (c *CaptiveStellarCore) GetLatestLedgerSequence() (uint32, error) { | ||
| c.mutex.Lock() | ||
| defer c.mutex.Unlock() | ||
|
|
||
| if c.isClosed() { | ||
| return 0, errors.New("stellar-core must be opened to return latest available sequence") | ||
| } | ||
| @@ -544,23 +589,27 @@ func (c *CaptiveStellarCore) isClosed() bool { | ||
| } | ||
|
|
||
| // Close closes existing Stellar-Core process, streaming sessions and removes all | ||
| // temporary files. Please note that this can be executed concurently, however it | ||
| // may block if `GetLedger` is running on a unbounded range (because `GetLedger` | ||
| // is blocking in this case). | ||
| func (c *CaptiveStellarCore) Close() error { | ||
| c.mutex.Lock() | ||
| defer c.mutex.Unlock() | ||
| return c.close() | ||
| } | ||
|
|
||
| func (c *CaptiveStellarCore) close() error { | ||
| if c.stellarCoreRunner != nil { | ||
| // Closing stellarCoreRunner will automatically close bufferedLedgerMetaReader's | ||
| // start() go routine because it's checking it's state. | ||
| err := c.stellarCoreRunner.close() | ||
| if err != nil { | ||
| return errors.Wrap(err, "error closing stellar-core subprocess") | ||
| } | ||
| } | ||
|
|
||
| if c.ledgerBuffer != nil { | ||
| c.ledgerBuffer.close() | ||
| } | ||
|
|
||
| c.nextLedger = 0 | ||
| @@ -569,11 +618,3 @@ func (c *CaptiveStellarCore) Close() error { | ||
|
|
||
| return nil | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Dyingis whenKillwas called but it's possible thattombis notDeadyet. We can use this signal to start shutting down go routine.