Repository: tablelandnetwork/go-tableland
Pull request #255 (2022-08-30 21:30:33)

executor: improve closing
The `Executor` had two missing things:
1. Close the db when `Close()` is called. (I wonder if this contributed to WAL prunning checkpointing delays)
2. There was a subtle race between `Close()` and getting a lock to enforce serial execution (will be explained in comments).


Patch:
diff --git a/.github/workflows/issues-notion-sync.yml b/.github/workflows/issues-notion-sync.yml
new file mode 100644
index 0000000..0a3c9ba
--- /dev/null
+++ b/.github/workflows/issues-notion-sync.yml
@@ -0,0 +1,29 @@
+name: Notion Sync
+
+on:
+  workflow_dispatch:
+  issues:
+    types:
+      [
+        opened,
+        edited,
+        labeled,
+        unlabeled,
+        assigned,
+        unassigned,
+        milestoned,
+        demilestoned,
+        reopened,
+        closed,
+      ]
+
+jobs:
+  notion_job:
+    runs-on: ubuntu-latest
+    name: Add GitHub Issues to Notion
+    steps:
+      - name: Add GitHub Issues to Notion
+        uses: tryfabric/notion-github-action@v1.2.1
+        with:
+          notion-token: ${{ secrets.NOTION_TOKEN }}
+          notion-db: ${{ secrets.NOTION_DATABASE }}
diff --git a/pkg/eventprocessor/eventfeed/impl/analytics.go b/pkg/eventprocessor/eventfeed/impl/analytics.go
index 4f61e15..fcdb6a7 100644
--- a/pkg/eventprocessor/eventfeed/impl/analytics.go
+++ b/pkg/eventprocessor/eventfeed/impl/analytics.go
@@ -9,7 +9,6 @@ import (
 var fetchBlockExtraInfoDelay = time.Second * 10
 
 func (ef *EventFeed) fetchExtraBlockInfo(ctx context.Context) {
-	var fromHeight *int64
 	for {
 		time.Sleep(fetchBlockExtraInfoDelay)
 
@@ -17,7 +16,7 @@ func (ef *EventFeed) fetchExtraBlockInfo(ctx context.Context) {
 			ef.log.Info().Msg("graceful close of extra block info fetcher")
 			return
 		}
-		blockNumbers, err := ef.systemStore.GetBlocksMissingExtraInfo(ctx, fromHeight)
+		blockNumbers, err := ef.systemStore.GetBlocksMissingExtraInfo(ctx)
 		if err != nil {
 			ef.log.Error().Err(err).Msg("get blocks without extra info")
 			continue
@@ -49,8 +48,5 @@ func (ef *EventFeed) fetchExtraBlockInfo(ctx context.Context) {
 		for i := 0; i < cap(rateLim); i++ {
 			rateLim <- struct{}{}
 		}
-		if len(blockNumbers) > 0 {
-			fromHeight = &blockNumbers[len(blockNumbers)-1]
-		}
 	}
 }
diff --git a/pkg/eventprocessor/impl/executor/impl/executor.go b/pkg/eventprocessor/impl/executor/impl/executor.go
index 3cdcbbb..479a33a 100644
--- a/pkg/eventprocessor/impl/executor/impl/executor.go
+++ b/pkg/eventprocessor/impl/executor/impl/executor.go
@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"sync"
 
 	"github.com/XSAM/otelsql"
 	"github.com/mattn/go-sqlite3"
@@ -26,6 +27,9 @@ type Executor struct {
 
 	chainID          tableland.ChainID
 	maxTableRowCount int
+
+	closeOnce sync.Once
+	closed    chan struct{}
 }
 
 var _ executor.Executor = (*Executor)(nil)
@@ -70,6 +74,8 @@ func NewExecutor(
 
 		chainID:          chainID,
 		maxTableRowCount: maxTableRowCount,
+
+		closed: make(chan struct{}),
 	}
 	tblp.chBlockScope <- struct{}{}
 
@@ -80,6 +86,8 @@ func NewExecutor(
 func (ex *Executor) NewBlockScope(ctx context.Context, newBlockNum int64) (executor.BlockScope, error) {
 	select {
 	case <-ex.chBlockScope:
+	case <-ex.closed:
+		return nil, fmt.Errorf("executor is closed")
 	default:
 		panic("parallel block scope detected, this must never happen")
 	}
@@ -138,10 +146,17 @@ func (ex *Executor) getLastExecutedBlockNumber(ctx context.Context, txn *sql.Tx)
 // Close closes the processor gracefully. It will wait for any pending
 // batch to be closed, or until ctx is canceled.
 func (ex *Executor) Close(ctx context.Context) error {
+	ex.closeOnce.Do(func() { close(ex.closed) })
 	select {
 	case <-ctx.Done():
+		if err := ex.db.Close(); err != nil {
+			ex.log.Error().Err(err).Msg("forced close of database connection")
+		}
 		return errors.New("closing ctx done")
 	case <-ex.chBlockScope:
+		if err := ex.db.Close(); err != nil {
+			ex.log.Error().Err(err).Msg("closing database connection")
+		}
 		ex.log.Info().Msg("executor closed gracefully")
 		return nil
 	}
diff --git a/pkg/sqlstore/impl/system/internal/db/evm_events.go b/pkg/sqlstore/impl/system/internal/db/evm_events.go
index e7c59fb..49b6574 100644
--- a/pkg/sqlstore/impl/system/internal/db/evm_events.go
+++ b/pkg/sqlstore/impl/system/internal/db/evm_events.go
@@ -102,18 +102,16 @@ func (q *Queries) AreEVMEventsPersisted(ctx context.Context, arg AreEVMTxnEvents
 const getBlocksMissingExtraInfo = `
 SELECT DISTINCT e.block_number
 FROM system_evm_events e 
-WHERE e.chain_id=?1 AND 
-      (?2 is null OR e.block_number>?2) AND
-      NOT EXISTS(select* from system_evm_blocks b WHERE e.chain_id=b.chain_id AND e.block_number=b.block_number)
+LEFT OUTER JOIN system_evm_blocks b ON e.chain_id=b.chain_id AND e.block_number=b.block_number
+WHERE e.chain_id=?1 AND b.block_number is null
 ORDER BY e.block_number`
 
 type GetBlocksMissingExtraInfoParams struct {
-	ChainID    int64
-	FromHeight *int64
+	ChainID int64
 }
 
 func (q *Queries) GetBlocksMissingExtraInfo(ctx context.Context, arg GetBlocksMissingExtraInfoParams) ([]int64, error) {
-	rows, err := q.query(ctx, q.getBlocksMissingExtraInfoStmt, getBlocksMissingExtraInfo, arg.ChainID, arg.FromHeight)
+	rows, err := q.query(ctx, q.getBlocksMissingExtraInfoStmt, getBlocksMissingExtraInfo, arg.ChainID)
 	if err != nil {
 		return nil, fmt.Errorf("executing getBlocksMissingExtraInfo query: %s", err)
 	}
diff --git a/pkg/sqlstore/impl/system/migrations/migrations.go b/pkg/sqlstore/impl/system/migrations/migrations.go
index 82d0206..aeaf09d 100644
--- a/pkg/sqlstore/impl/system/migrations/migrations.go
+++ b/pkg/sqlstore/impl/system/migrations/migrations.go
@@ -1,6 +1,6 @@
 // Code generated by go-bindata. (@generated) DO NOT EDIT.
 
- //Package migrations generated by go-bindata.// sources:
+//Package migrations generated by go-bindata.// sources:
 // migrations/001_init.down.sql
 // migrations/001_init.up.sql
 // migrations/002_receipterroridx.down.sql
@@ -178,7 +178,7 @@ func _003_evm_eventsDownSql() (*asset, error) {
 		return nil, err
 	}
 
-	info := bindataFileInfo{name: "003_evm_events.down.sql", size: 59, mode: os.FileMode(436), modTime: time.Unix(1661377601, 0)}
+	info := bindataFileInfo{name: "003_evm_events.down.sql", size: 59, mode: os.FileMode(436), modTime: time.Unix(1661268420, 0)}
 	a := &asset{bytes: bytes, info: info}
 	return a, nil
 }
@@ -198,7 +198,7 @@ func _003_evm_eventsUpSql() (*asset, error) {
 		return nil, err
 	}
 
-	info := bindataFileInfo{name: "003_evm_events.up.sql", size: 701, mode: os.FileMode(436), modTime: time.Unix(1661377601, 0)}
+	info := bindataFileInfo{name: "003_evm_events.up.sql", size: 701, mode: os.FileMode(436), modTime: time.Unix(1661268422, 0)}
 	a := &asset{bytes: bytes, info: info}
 	return a, nil
 }
diff --git a/pkg/sqlstore/impl/system/store.go b/pkg/sqlstore/impl/system/store.go
index cae0551..86a2a60 100644
--- a/pkg/sqlstore/impl/system/store.go
+++ b/pkg/sqlstore/impl/system/store.go
@@ -376,12 +376,9 @@ func (s *SystemStore) SaveEVMEvents(ctx context.Context, events []tableland.EVME
 }
 
 // GetBlocksMissingExtraInfo returns a list of block numbers that don't contain enhanced information.
-// It receives an optional fromHeight to only look for blocks after a block number. If null it will look
-// for blocks at any height.
-func (s *SystemStore) GetBlocksMissingExtraInfo(ctx context.Context, lastKnownHeight *int64) ([]int64, error) {
+func (s *SystemStore) GetBlocksMissingExtraInfo(ctx context.Context) ([]int64, error) {
 	params := db.GetBlocksMissingExtraInfoParams{
-		ChainID:    int64(s.chainID),
-		FromHeight: lastKnownHeight,
+		ChainID: int64(s.chainID),
 	}
 	blockNumbers, err := s.dbWithTx.queries().GetBlocksMissingExtraInfo(ctx, params)
 	if err == sql.ErrNoRows {
diff --git a/pkg/sqlstore/impl/system_store_instrumented.go b/pkg/sqlstore/impl/system_store_instrumented.go
index ec07f91..3412ad0 100644
--- a/pkg/sqlstore/impl/system_store_instrumented.go
+++ b/pkg/sqlstore/impl/system_store_instrumented.go
@@ -333,13 +333,10 @@ func (s *InstrumentedSystemStore) GetEVMEvents(ctx context.Context, txnHash comm
 }
 
 // GetBlocksMissingExtraInfo implements sqlstore.SystemStore.
-func (s *InstrumentedSystemStore) GetBlocksMissingExtraInfo(
-	ctx context.Context,
-	fromHeight *int64,
-) ([]int64, error) {
+func (s *InstrumentedSystemStore) GetBlocksMissingExtraInfo(ctx context.Context) ([]int64, error) {
 	log.Debug().Msg("call GetBlocksMissingExtraInfo")
 	start := time.Now()
-	blockNumbers, err := s.store.GetBlocksMissingExtraInfo(ctx, fromHeight)
+	blockNumbers, err := s.store.GetBlocksMissingExtraInfo(ctx)
 	latency := time.Since(start).Milliseconds()
 
 	attributes := []attribute.KeyValue{
diff --git a/pkg/sqlstore/store_system.go b/pkg/sqlstore/store_system.go
index ebe3a71..04d82e1 100644
--- a/pkg/sqlstore/store_system.go
+++ b/pkg/sqlstore/store_system.go
@@ -30,7 +30,7 @@ type SystemStore interface {
 	AreEVMEventsPersisted(context.Context, common.Hash) (bool, error)
 	SaveEVMEvents(context.Context, []tableland.EVMEvent) error
 	GetEVMEvents(context.Context, common.Hash) ([]tableland.EVMEvent, error)
-	GetBlocksMissingExtraInfo(context.Context, *int64) ([]int64, error)
+	GetBlocksMissingExtraInfo(context.Context) ([]int64, error)
 	InsertBlockExtraInfo(context.Context, int64, uint64) error
 	GetBlockExtraInfo(context.Context, int64) (tableland.EVMBlockInfo, error)
 


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index a8916c0..89412d2 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -14,7 +14,7 @@ jobs:
   test:
     strategy:
       matrix:
-        go-version: [1.18.x, v1.19.x]
+        go-version: [1.17.x, v1.18.x]
         os: [ubuntu-latest]
     runs-on: ${{ matrix.os }}
     steps: