core/bloombits, eth/filters: bubble up errors

core/bloombits/matcher.go

@@ -60,6 +60,7 @@ type Retrieval struct {
 	Bit      uint
 	Sections []uint64
 	Bitsets  [][]byte
+	Error    error
 }
 
 // Matcher is a pipelined system of schedulers and logic matchers which perform
@@ -502,15 +503,27 @@ func (m *Matcher) distributor(dist chan *request, session *MatcherSession) {
 type MatcherSession struct {
 	matcher *Matcher
 
-	quit chan struct{} // Quit channel to request pipeline termination
+	quit     chan struct{} // Quit channel to request pipeline termination
-	kill chan struct{} // Term channel to signal non-graceful forced shutdown
+	kill     chan struct{} // Term channel to signal non-graceful forced shutdown
-	pend sync.WaitGroup
+	err      error
+	stopping bool
+	lock     sync.Mutex
+	pend     sync.WaitGroup
 }
 
 // Close stops the matching process and waits for all subprocesses to terminate
 // before returning. The timeout may be used for graceful shutdown, allowing the
 // currently running retrievals to complete before this time.
-func (s *MatcherSession) Close(timeout time.Duration) {
+func (s *MatcherSession) Close() {
+	s.lock.Lock()
+	stopping := s.stopping
+	s.stopping = true
+	s.lock.Unlock()
+	// ensure that we only close the session once
+	if stopping {
+		return
+	}
+
 	// Bail out if the matcher is not running
 	select {
 	case <-s.quit:
@@ -519,10 +532,26 @@ func (s *MatcherSession) Close(timeout time.Duration) {
 	}
 	// Signal termination and wait for all goroutines to tear down
 	close(s.quit)
-	time.AfterFunc(timeout, func() { close(s.kill) })
+	time.AfterFunc(time.Second, func() { close(s.kill) })
 	s.pend.Wait()
 }
 
+// setError sets an error and stops the session
+func (s *MatcherSession) setError(err error) {
+	s.lock.Lock()
+	s.err = err
+	s.lock.Unlock()
+	s.Close()
+}
+
+// Error returns an error if one has happened during the session
+func (s *MatcherSession) Error() error {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	return s.err
+}
+
 // AllocateRetrieval assigns a bloom bit index to a client process that can either
 // immediately reuest and fetch the section contents assigned to this bit or wait
 // a little while for more sections to be requested.
@@ -621,6 +650,10 @@ func (s *MatcherSession) Multiplex(batch int, wait time.Duration, mux chan chan
 			request <- &Retrieval{Bit: bit, Sections: sections}
 
 			result := <-request
+			if result.Error != nil {
+				s.setError(result.Error)
+			}
+
 			s.DeliverSections(result.Bit, result.Sections, result.Bitsets)
 		}
 	}

core/bloombits/matcher_test.go

@@ -163,7 +163,7 @@ func testMatcher(t *testing.T, filter [][]bloomIndexes, blocks uint64, intermitt
 			}
 			// If we're testing intermittent mode, abort and restart the pipeline
 			if intermittent {
-				session.Close(time.Second)
+				session.Close()
 				close(quit)
 
 				quit = make(chan struct{})
@@ -183,7 +183,7 @@ func testMatcher(t *testing.T, filter [][]bloomIndexes, blocks uint64, intermitt
 		t.Errorf("filter = %v  blocks = %v  intermittent = %v: expected closed channel, got #%v", filter, blocks, intermittent, match)
 	}
 	// Clean up the session and ensure we match the expected retrieval count
-	session.Close(time.Second)
+	session.Close()
 	close(quit)
 
 	if retrievals != 0 && requested != retrievals {

core/database_util.go

@@ -332,14 +332,13 @@ func GetReceipt(db DatabaseReader, hash common.Hash) (*types.Receipt, common.Has
 
 // GetBloomBits retrieves the compressed bloom bit vector belonging to the given
 // section and bit index from the.
-func GetBloomBits(db DatabaseReader, bit uint, section uint64, head common.Hash) []byte {
+func GetBloomBits(db DatabaseReader, bit uint, section uint64, head common.Hash) ([]byte, error) {
 	key := append(append(bloomBitsPrefix, make([]byte, 10)...), head.Bytes()...)
 
 	binary.BigEndian.PutUint16(key[1:], uint16(bit))
 	binary.BigEndian.PutUint64(key[3:], section)
 
-	bits, _ := db.Get(key)
+	return db.Get(key)
-	return bits
 }
 
 // WriteCanonicalHash stores the canonical hash for the given block number.

eth/bloombits.go

@@ -58,15 +58,18 @@ func (eth *Ethereum) startBloomHandlers() {
 
 				case request := <-eth.bloomRequests:
 					task := <-request
-
 					task.Bitsets = make([][]byte, len(task.Sections))
 					for i, section := range task.Sections {
 						head := core.GetCanonicalHash(eth.chainDb, (section+1)*params.BloomBitsBlocks-1)
-						blob, err := bitutil.DecompressBytes(core.GetBloomBits(eth.chainDb, task.Bit, section, head), int(params.BloomBitsBlocks)/8)
+						if compVector, err := core.GetBloomBits(eth.chainDb, task.Bit, section, head); err == nil {
-						if err != nil {
+							if blob, err := bitutil.DecompressBytes(compVector, int(params.BloomBitsBlocks)/8); err == nil {
-							panic(err)
+								task.Bitsets[i] = blob
+							} else {
+								task.Error = err
+							}
+						} else {
+							task.Error = err
 						}
-						task.Bitsets[i] = blob
 					}
 					request <- task
 				}

eth/filters/filter.go

@@ -19,7 +19,6 @@ package filters
 import (
 	"context"
 	"math/big"
-	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
@@ -140,7 +139,7 @@ func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, err
 	if err != nil {
 		return nil, err
 	}
-	defer session.Close(time.Second)
+	defer session.Close()
 
 	f.backend.ServiceFilter(ctx, session)
 
@@ -152,9 +151,13 @@ func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, err
 		case number, ok := <-matches:
 			// Abort if all matches have been fulfilled
 			if !ok {
-				f.begin = int64(end) + 1
+				err := session.Error()
-				return logs, nil
+				if err == nil {
+					f.begin = int64(end) + 1
+				}
+				return logs, err
 			}
+			f.begin = int64(number) + 1
 			// Retrieve the suggested block and pull any truly matching logs
 			header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(number))
 			if header == nil || err != nil {

eth/filters/filter_system_test.go

@@ -109,7 +109,7 @@ func (b *testBackend) ServiceFilter(ctx context.Context, session *bloombits.Matc
 				for i, section := range task.Sections {
 					if rand.Int()%4 != 0 { // Handle occasional missing deliveries
 						head := core.GetCanonicalHash(b.db, (section+1)*params.BloomBitsBlocks-1)
-						task.Bitsets[i] = core.GetBloomBits(b.db, task.Bit, section, head)
+						task.Bitsets[i], _ = core.GetBloomBits(b.db, task.Bit, section, head)
 					}
 				}
 				request <- task