Repository: livepeer/go-livepeer
Pull request #2493 (2022-07-11 09:22:41)

Fix reward flag
**What does this pull request do? Explain your changes. (required)**
Fix a bug in the `reward` flag. It stopped working after the refactoring PR #2351. If the flag is set explicitly to `true`, it does not work. This PR fixes it.

**Specific updates (required)**
<!--- List out all significant updates your code introduces -->
- 
- 
- 

**How did you test each of these updates (required)**
<!-- A detailed description of how you tested your code changes. Include details of your testing environment, and the tests you ran to see how your change affects other areas of the code, etc. -->
Tested manually

**Does this pull request close any open issues?**
<!-- Fixes # -->
N/A


**Checklist:**
<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->
- [x] Read the [contribution guide](./doc/contributing.md)
- [x] `make` runs successfully
- [x] All tests in `./test.sh` pass
- [ ] ~README and other documentation updated~
- [x] [Pending changelog](./CHANGELOG_PENDING.md) updated


codecov[bot]: # [Codecov](https://codecov.io/gh/livepeer/go-livepeer/pull/2493?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=livepeer) Report
> Merging [#2493](https://codecov.io/gh/livepeer/go-livepeer/pull/2493?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=livepeer) (426d7ef) into [master](https://codecov.io/gh/livepeer/go-livepeer/commit/5ad8ab5e98d0bed48c62fa760231befe5debf6c1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=livepeer) (5ad8ab5) will **decrease** coverage by `0.02338%`.
> The diff coverage is `0.00000%`.

> :exclamation: Current head 426d7ef differs from pull request most recent head 4604005. Consider uploading reports for the commit 4604005 to get more accurate results

[![Impacted file tree graph](https://codecov.io/gh/livepeer/go-livepeer/pull/2493/graphs/tree.svg?width=650&height=150&src=pr&token=vKzl2rUYvB&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=livepeer)](https://codecov.io/gh/livepeer/go-livepeer/pull/2493?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=livepeer)

```diff
@@                 Coverage Diff                 @@
##              master       #2493         +/-   ##
===================================================
- Coverage   54.73000%   54.70662%   -0.02338%     
===================================================
  Files             95          95                 
  Lines          19852       19855          +3     
===================================================
- Hits           10865       10862          -3     
- Misses          8383        8389          +6     
  Partials         604         604                 
```


| [Impacted Files](https://codecov.io/gh/livepeer/go-livepeer/pull/2493?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=livepeer) | Coverage Δ | |
|---|---|---|
| [cmd/livepeer/starter/starter.go](https://codecov.io/gh/livepeer/go-livepeer/pull/2493/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=livepeer#diff-Y21kL2xpdmVwZWVyL3N0YXJ0ZXIvc3RhcnRlci5nbw==) | `2.29665% <0.00000%> (-0.00661%)` | :arrow_down: |
| [discovery/db\_discovery.go](https://codecov.io/gh/livepeer/go-livepeer/pull/2493/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=livepeer#diff-ZGlzY292ZXJ5L2RiX2Rpc2NvdmVyeS5nbw==) | `67.01389% <0.00000%> (-1.04167%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/livepeer/go-livepeer/pull/2493?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=livepeer).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=livepeer)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/livepeer/go-livepeer/pull/2493?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=livepeer). Last update [5ad8ab5...4604005](https://codecov.io/gh/livepeer/go-livepeer/pull/2493?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=livepeer). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=livepeer).


Patch:
diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md
index 8eff160a..96736eae 100644
--- a/CHANGELOG_PENDING.md
+++ b/CHANGELOG_PENDING.md
@@ -9,6 +9,7 @@
 #### General
 - \#1333 Display git-sha in startup logging (@emranemran)
 - \#2443 Add e2e tests for O configuration and resignation (@red-0ne)
+- \#2489 Backfill blocks in batches (@leszko)
 
 #### Broadcaster
 - \#2462 cmd: Delete temporary env variable LP_IS_ORCH_TESTER (@leszko)
@@ -32,5 +33,6 @@
 #### Broadcaster
 
 #### Orchestrator
+- \#2493 cmd: Fix reward flag (@leszko)
 
 #### Transcoder
diff --git a/cmd/livepeer/starter/starter.go b/cmd/livepeer/starter/starter.go
index 8bd9106b..22e879fa 100755
--- a/cmd/livepeer/starter/starter.go
+++ b/cmd/livepeer/starter/starter.go
@@ -846,6 +846,8 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 			} else {
 				reward = false
 			}
+		} else {
+			reward = *cfg.Reward
 		}
 
 		if reward {
@@ -879,10 +881,12 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 
 		// Backfill events that the node has missed since its last seen block. This method will block
 		// and the node will not continue setup until it finishes
-		if err := blockWatcher.BackfillEventsIfNeeded(blockWatchCtx); err != nil {
+		glog.Infof("Backfilling block events (this can take a while)...\n")
+		if err := blockWatcher.BackfillEvents(blockWatchCtx); err != nil {
 			glog.Errorf("Failed to backfill events: %v", err)
 			return
 		}
+		glog.Info("Done backfilling block events")
 
 		blockWatcherErr := make(chan error, 1)
 		go func() {
diff --git a/eth/blockwatch/block_watcher.go b/eth/blockwatch/block_watcher.go
index c097b580..bc932776 100644
--- a/eth/blockwatch/block_watcher.go
+++ b/eth/blockwatch/block_watcher.go
@@ -85,10 +85,12 @@ func New(config Config) *Watcher {
 	return bs
 }
 
-// BackfillEventsIfNeeded finds missed events that might have occured while the
-// node was offline and sends them to event subscribers. It blocks until
-// it is done backfilling or the given context is canceled.
-func (w *Watcher) BackfillEventsIfNeeded(ctx context.Context) error {
+// BackfillEvents finds missed events and sends them to event subscribers.
+// It blocks until it is done backfilling or the given context is canceled.
+// Note that the latest block is never backfilled here from logs. It will be polled separately in syncToLatestBlock().
+// The reason for that is that we always need to propagate events from the latest block even if it does not contain
+// events which are filtered out during the backfilling process.
+func (w *Watcher) BackfillEvents(ctx context.Context) error {
 	events, err := w.getMissedEventsToBackfill(ctx)
 	if err != nil {
 		return err
@@ -117,7 +119,7 @@ func (w *Watcher) Watch(ctx context.Context) error {
 			ticker.Stop()
 			return nil
 		case <-ticker.C:
-			if err := w.syncToLatestBlock(); err != nil {
+			if err := w.syncToLatestBlock(ctx); err != nil {
 				glog.Errorf("blockwatch.Watcher error encountered - trying again on next polling interval err=%q", err)
 			}
 		}
@@ -147,9 +149,14 @@ func (w *Watcher) InspectRetainedBlocks() ([]*MiniHeader, error) {
 	return w.stack.Inspect()
 }
 
-func (w *Watcher) syncToLatestBlock() error {
+func (w *Watcher) syncToLatestBlock(ctx context.Context) error {
 	w.Lock()
 	defer w.Unlock()
+
+	if err := w.BackfillEvents(ctx); err != nil {
+		return err
+	}
+
 	newestHeader, err := w.client.HeaderByNumber(nil)
 	if err != nil {
 		return err
@@ -303,10 +310,11 @@ func (w *Watcher) addLogs(header *MiniHeader) (*MiniHeader, error) {
 	return header, nil
 }
 
-// getMissedEventsToBackfill finds missed events that might have occured while the node was
-// offline. It does this by comparing the last block stored with the latest block discoverable via RPC.
-// If the stored block is older then the latest block, it batch fetches the events for missing blocks,
+// getMissedEventsToBackfill finds missed events that might have occurred since the last block polling.
+// It does this by comparing the last block stored with the latest block discoverable via RPC.
+// If the stored block is older than the latest block, it batch-fetches the events for missing blocks,
 // re-sets the stored blocks and returns the block events found.
+// Note that the latest block is never backfilled, and will be polled separately during in syncToLatestBlock().
 func (w *Watcher) getMissedEventsToBackfill(ctx context.Context) ([]*Event, error) {
 	events := []*Event{}
 
@@ -325,7 +333,9 @@ func (w *Watcher) getMissedEventsToBackfill(ctx context.Context) ([]*Event, erro
 	if err != nil {
 		return events, err
 	}
-	latestBlockNum := int(latestBlock.Number.Int64())
+
+	// Latest block will be polled separately in syncToLatestBlock(), so it's not backfilled.
+	preLatestBlockNum := int(latestBlock.Number.Int64()) - 1
 
 	if latestRetainedBlock != nil {
 		latestRetainedBlockNum = int(latestRetainedBlock.Number.Int64())
@@ -335,14 +345,11 @@ func (w *Watcher) getMissedEventsToBackfill(ctx context.Context) ([]*Event, erro
 		return events, nil
 	}
 
-	if blocksElapsed = latestBlockNum - startBlockNum; blocksElapsed <= 0 {
+	if blocksElapsed = preLatestBlockNum - startBlockNum; blocksElapsed <= 0 {
 		return events, nil
 	}
 
-	glog.Infof("Backfilling block events (this can take a while)...\n")
-	glog.Infof("Start block: %v		 End block: %v		Blocks elapsed: %v\n", startBlockNum, startBlockNum+blocksElapsed, blocksElapsed)
-
-	logs, furthestBlockProcessed := w.getLogsInBlockRange(ctx, startBlockNum, latestBlockNum)
+	logs, furthestBlockProcessed := w.getLogsInBlockRange(ctx, startBlockNum, preLatestBlockNum)
 	if furthestBlockProcessed > latestRetainedBlockNum {
 		// If we have processed blocks further then the latestRetainedBlock in the DB, we
 		// want to remove all blocks from the DB and insert the furthestBlockProcessed
@@ -396,7 +403,6 @@ func (w *Watcher) getMissedEventsToBackfill(ctx context.Context) ([]*Event, erro
 				BlockHeader: blockHeader,
 			})
 		}
-		glog.Info("Done backfilling block events")
 		return events, nil
 	}
 	return events, nil
@@ -558,7 +564,7 @@ func (w *Watcher) getSubBlockRanges(from, to, rangeSize int) []*blockRange {
 const infuraTooManyResultsErrMsg = "query returned more than 10000 results"
 
 func (w *Watcher) filterLogsRecursively(from, to int, allLogs []types.Log) ([]types.Log, error) {
-	glog.Infof("fetching block logs from=%v to=%v", from, to)
+	glog.V(6).Infof("Polling blocks from=%v to=%v", from, to)
 	numBlocks := to - from
 	topics := [][]common.Hash{}
 	if len(w.topics) > 0 {


Test patch:
diff --git a/eth/blockwatch/block_watcher_test.go b/eth/blockwatch/block_watcher_test.go
index f94f68b4..47ac1374 100644
--- a/eth/blockwatch/block_watcher_test.go
+++ b/eth/blockwatch/block_watcher_test.go
@@ -189,12 +189,13 @@ func TestGetMissedEventsToBackfillSomeMissed(t *testing.T) {
 
 	store := &stubMiniHeaderStore{}
 	// Add block number 5 as the last block seen by BlockWatcher
-	lastBlockSeen := &MiniHeader{
+	preLastBlockSeen := &MiniHeader{
 		Number: big.NewInt(5),
 		Hash:   common.HexToHash("0x293b9ea024055a3e9eddbf9b9383dc7731744111894af6aa038594dc1b61f87f"),
 		Parent: common.HexToHash("0x26b13ac89500f7fcdd141b7d1b30f3a82178431eca325d1cf10998f9d68ff5ba"),
 	}
-	err = store.InsertMiniHeader(lastBlockSeen)
+	err = store.InsertMiniHeader(preLastBlockSeen)
+
 	require.NoError(t, err)
 
 	config.Store = store
@@ -211,7 +212,7 @@ func TestGetMissedEventsToBackfillSomeMissed(t *testing.T) {
 	headers, err := store.FindAllMiniHeadersSortedByNumber()
 	require.NoError(t, err)
 	require.Len(t, headers, 1)
-	assert.Equal(t, big.NewInt(30), headers[0].Number)
+	assert.Equal(t, big.NewInt(29), headers[0].Number)
 }
 
 func TestGetMissedEventsToBackfillNoneMissed(t *testing.T) {
diff --git a/eth/blockwatch/testdata/fake_client_fast_sync_fixture.json b/eth/blockwatch/testdata/fake_client_fast_sync_fixture.json
index 2e57c5cc..36c4149f 100644
--- a/eth/blockwatch/testdata/fake_client_fast_sync_fixture.json
+++ b/eth/blockwatch/testdata/fake_client_fast_sync_fixture.json
@@ -15,6 +15,11 @@
             "hash": "0x382b25dd926322722bba2575b5092be661a99f26472e2b7c2bb3d51e6bd2b09c",
             "parent": "0xb57233b65a0da953da19d685d10a65e3124207fd1cf4694e3fc0f7279373bf5f",
             "number": 30
+          },
+          "29": {
+            "hash": "0xb57233b65a0da953da19d685d10a65e3124207fd1cf4694e3fc0f7279373bf5f",
+            "parent": "0x0f52679a1ff257072b0588234c9cf9727cc4fe3a973e5fa528474d2be638253b",
+            "number": 29
           }
         },
         "getBlockByHash": {
@@ -27,6 +32,11 @@
             "hash": "0x382b25dd926322722bba2575b5092be661a99f26472e2b7c2bb3d51e6bd2b09c",
             "parent": "0xb57233b65a0da953da19d685d10a65e3124207fd1cf4694e3fc0f7279373bf5f",
             "number": 30
+          },
+          "0xb57233b65a0da953da19d685d10a65e3124207fd1cf4694e3fc0f7279373bf5f": {
+            "hash": "0xb57233b65a0da953da19d685d10a65e3124207fd1cf4694e3fc0f7279373bf5f",
+            "parent": "0x0f52679a1ff257072b0588234c9cf9727cc4fe3a973e5fa528474d2be638253b",
+            "number": 29
           }
         },
         "getCorrectChain": [],