Fix dbstore iterator bug and add even more logging

p2p/protocols/protocol.go

@@ -34,6 +34,7 @@ import (
 	"reflect"
 	"sync"
 
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
 )
 
@@ -210,6 +211,7 @@ func (p *Peer) Run(handler func(msg interface{}) error) error {
 // if they are useful for other protocols
 // overwrite Disconnect for testing, so that protocol readloop quits
 func (p *Peer) Drop(err error) {
+	log.Error("p2p protocol DROP", "err", err)
 	p.Disconnect(p2p.DiscSubprotocolError)
 }
 

swarm/network/stream/delivery.go

@@ -19,7 +19,6 @@ package stream
 import (
 	"errors"
 	"fmt"
-	"os"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -100,9 +99,14 @@ func (s *SwarmChunkServer) SetNextBatch(_, _ uint64) (hashes []byte, from uint64
 }
 
 // GetData retrives chunk data from db store
-func (s *SwarmChunkServer) GetData(key []byte) []byte {
-	chunk, _ := s.db.Get(storage.Key(key))
-	return chunk.SData
+func (s *SwarmChunkServer) GetData(key []byte) ([]byte, error) {
+	chunk, err := s.db.Get(storage.Key(key))
+	if err == storage.ErrFetching {
+		<-chunk.ReqC
+	} else if err != nil {
+		return nil, err
+	}
+	return chunk.SData, nil
 }
 
 // RetrieveRequestMsg is the protocol msg for chunk retrieve requests
@@ -141,7 +145,7 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e
 			if req.SkipCheck {
 				err := sp.Deliver(chunk, s.priority)
 				if err != nil {
-					sp.Drop(err)
+					sp.Drop(fmt.Errorf("handleRetrieveRequestMsg: %v", err))
 				}
 			}
 			streamer.deliveryC <- chunk.Key[:]
@@ -173,9 +177,9 @@ R:
 	for req := range d.receiveC {
 		// this should be has locally
 		chunk, err := d.db.Get(req.Key)
-		fmt.Fprintln(os.Stderr, "pick from receiveC", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err)
+		log.Error("pick from receiveC", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err)
 		if err == nil {
-			fmt.Fprintln(os.Stderr, "found existing?", "hash", chunk.Key.Hex())
+			log.Error("found existing?", "hash", chunk.Key.Hex())
 			continue R
 		}
 		if err != storage.ErrFetching {
@@ -183,19 +187,19 @@ R:
 		}
 		select {
 		case <-chunk.ReqC:
-			fmt.Fprintln(os.Stderr, "someone else delivered?", "hash", chunk.Key.Hex())
+			log.Error("someone else delivered?", "hash", chunk.Key.Hex())
 			continue R
 		default:
 		}
 		go func() {
 			chunk.SData = req.SData
-			fmt.Fprintln(os.Stderr, "received delivery", "hash", chunk.Key.Hex())
+			log.Error("received delivery", "hash", chunk.Key.Hex())
 			d.db.Put(chunk)
-			fmt.Fprintln(os.Stderr, "put to db", "hash", chunk.Key.Hex())
+			log.Error("put to db", "hash", chunk.Key.Hex())
 			chunk.WaitToStore()
 			close(chunk.ReqC)
 			//log.Warn("received delivery stored", "hash", chunk.Key)
-			fmt.Fprintln(os.Stderr, "requesters notified", "hash", chunk.Key.Hex())
+			log.Error("requesters notified", "hash", chunk.Key.Hex())
 			d.counterDone++
 		}()
 	}

swarm/network/stream/delivery_test.go

@@ -380,7 +380,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 		// which responds to chunk retrieve requests all but the last node in the chain does not
 		var j int
 		err := sim.CallClient(func(client *rpc.Client) error {
-			err := streamTesting.WatchDisconnections(sim.IDs[j], client, errc, quitC)
+			err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC)
 			if err != nil {
 				return err
 			}
@@ -547,7 +547,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 		// which responds to chunk retrieve requests
 		var j int
 		simErrC <- sim.CallClient(func(client *rpc.Client) error {
-			err := streamTesting.WatchDisconnections(sim.IDs[j], client, simErrC, quitC)
+			err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), simErrC, quitC)
 			if err != nil {
 				return err
 			}

swarm/network/stream/messages.go

@@ -83,7 +83,7 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) error {
 	log.Debug("received subscription", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To)
 	go func() {
 		if err := p.SendOfferedHashes(os, req.From, req.To); err != nil {
-			p.Drop(err)
+			p.Drop(fmt.Errorf("handleSubscribeMsg SendOfferedHashes: %v", err))
 		}
 	}()
 	return nil
@@ -176,7 +176,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 			return
 		case err := <-s.next:
 			if err != nil {
-				p.Drop(err)
+				p.Drop(fmt.Errorf("handleOfferedHashesMsg next: %v", err))
 				return
 			}
 		case <-s.quit:
@@ -185,7 +185,7 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error {
 		log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "Key", msg.Key, "from", msg.From, "to", msg.To)
 		err := p.SendPriority(msg, s.priority)
 		if err != nil {
-			p.Drop(err)
+			p.Drop(fmt.Errorf("handleOfferedHashesMsg set priority: %v", err))
 		}
 	}()
 	return nil
@@ -218,7 +218,7 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error {
 	// launch in go routine since GetBatch blocks until new hashes arrive
 	go func() {
 		if err := p.SendOfferedHashes(s, req.From, req.To); err != nil {
-			p.Drop(err)
+			p.Drop(fmt.Errorf("handleWantedHashesMsg SendOfferedHashes: %v", err))
 		}
 	}()
 	// go p.SendOfferedHashes(s, req.From, req.To)
@@ -230,9 +230,9 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error {
 	for i := 0; i < l; i++ {
 		if want.Get(i) {
 			hash := hashes[i*HashSize : (i+1)*HashSize]
-			data := s.GetData(hash)
-			if data == nil {
-				return errors.New("not found")
+			data, err := s.GetData(hash)
+			if err != nil {
+				return fmt.Errorf("handleWantedHashesMsg get data %x: %v", hash, err)
 			}
 			chunk := storage.NewChunk(hash, nil)
 			chunk.SData = data

swarm/network/stream/stream.go

@@ -256,7 +256,7 @@ type server struct {
 // Server interface for outgoing peer Streamer
 type Server interface {
 	SetNextBatch(uint64, uint64) (hashes []byte, from uint64, to uint64, proof *HandoverProof, err error)
-	GetData([]byte) []byte
+	GetData([]byte) ([]byte, error)
 }
 
 type client struct {

swarm/network/stream/streamer_test.go

@@ -81,8 +81,8 @@ func (self *testServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, ui
 	return make([]byte, HashSize), from + 1, to + 1, nil, nil
 }
 
-func (self *testServer) GetData([]byte) []byte {
-	return nil
+func (self *testServer) GetData([]byte) ([]byte, error) {
+	return nil, nil
 }
 
 func TestStreamerDownstreamSubscribeMsgExchange(t *testing.T) {

swarm/network/stream/syncer.go

@@ -73,14 +73,14 @@ func RegisterSwarmSyncerServer(streamer *Registry, db *storage.DBAPI) {
 }
 
 // GetSection retrieves the actual chunk from localstore
-func (s *SwarmSyncerServer) GetData(key []byte) []byte {
+func (s *SwarmSyncerServer) GetData(key []byte) ([]byte, error) {
 	chunk, err := s.db.Get(storage.Key(key))
 	if err == storage.ErrFetching {
 		<-chunk.ReqC
 	} else if err != nil {
-		return nil
+		return nil, err
 	}
-	return chunk.SData
+	return chunk.SData, nil
 }
 
 // GetBatch retrieves the next batch of hashes from the dbstore
@@ -111,7 +111,7 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6
 	}
 
 	log.Debug("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.db.CurrentBucketStorageIndex(s.po))
-	return batch, from, to + 1, nil, nil
+	return batch, from, to, nil, nil
 }
 
 // SwarmSyncerClient

swarm/network/stream/syncer_test.go

@@ -22,7 +22,6 @@ import (
 	"fmt"
 	"io"
 	"math"
-	"os"
 	"testing"
 	"time"
 
@@ -142,7 +141,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 		// each node Subscribes to each other's swarmChunkServerStreamName
 		j := 0
 		return sim.CallClient(func(client *rpc.Client) error {
-			err := streamTesting.WatchDisconnections(sim.IDs[j], client, errc, quitC)
+			err := streamTesting.WatchDisconnections(sim.IDs[j], client, peerCount(sim.IDs[j]), errc, quitC)
 			if err != nil {
 				return err
 			}
@@ -182,7 +181,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 					} else if err == nil {
 						nodeHashFound++
 					} else {
-						fmt.Fprintln(os.Stderr, time.Now(), "not found", "index", i, "origin", j, "key", key.Hex(), "err", err)
+						log.Error("not found", "index", i, "origin", j, "key", key.Hex(), "err", err)
 					}
 				}
 			}

swarm/network/stream/testing/testing.go

@@ -154,9 +154,9 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) {
 	// set nodes number of Stores available
 	stores, storeTeardown, err := SetStores(addrs...)
 	teardown = func() {
-		storeTeardown()
-		adapterTeardown()
 		net.Shutdown()
+		adapterTeardown()
+		storeTeardown()
 	}
 	if err != nil {
 		return nil, teardown, err
@@ -208,7 +208,7 @@ func (s *Simulation) Run(ctx context.Context, conf *RunConfig) (*simulations.Ste
 	return result, nil
 }
 
-func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error, quitC chan struct{}) error {
+func WatchDisconnections(id discover.NodeID, client *rpc.Client, expectedConnCount int, errc chan error, quitC chan struct{}) error {
 	events := make(chan *p2p.PeerEvent)
 	sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents")
 	if err != nil {
@@ -218,10 +218,14 @@ func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error
 		defer sub.Unsubscribe()
 		select {
 		case <-quitC:
+			if expectedConnCount <= 0 {
 				return
+			}
 		case e := <-events:
+			expectedConnCount--
 			errc <- fmt.Errorf("peerEvent for node %v: %v", id, e)
 		case err := <-sub.Err():
+			expectedConnCount = 0
 			if err != nil {
 				errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err)
 			}

swarm/storage/dbstore.go

@@ -30,7 +30,6 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
-	"os"
 	"sync"
 	"time"
 
@@ -540,16 +539,16 @@ func (s *DbStore) CurrentStorageIndex() uint64 {
 }
 
 func (s *DbStore) Put(chunk *Chunk) {
-	fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put", "hash", chunk.Key.Hex())
+	log.Error("DbStore.Put", "hash", chunk.Key.Hex())
 	done := make(chan struct{})
 	defer close(done)
 	go func() {
-		fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put WAITER STARTED", "hash", chunk.Key.Hex())
+		log.Error("DbStore.Put WAITER STARTED", "hash", chunk.Key.Hex())
 		select {
 		case <-time.After(1 * time.Second):
-			fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put WAITING", "hash", chunk.Key.Hex())
+			log.Error("DbStore.Put WAITING", "hash", chunk.Key.Hex())
 		case <-done:
-			fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put EXITED", "hash", chunk.Key.Hex())
+			log.Error("DbStore.Put EXITED", "hash", chunk.Key.Hex())
 		}
 	}()
 
@@ -557,35 +556,35 @@ func (s *DbStore) Put(chunk *Chunk) {
 	var index dpaDBIndex
 
 	po := s.po(chunk.Key)
-	fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Get is being called...", "hash", chunk.Key.Hex())
+	log.Error("DbStore.db.Get is being called...", "hash", chunk.Key.Hex())
 
-	fmt.Fprintln(os.Stderr, time.Now(), "DbStore.LOCK acquiring", "hash", chunk.Key.Hex())
+	log.Error("DbStore.LOCK acquiring", "hash", chunk.Key.Hex())
 	s.lock.Lock()
-	fmt.Fprintln(os.Stderr, time.Now(), "DbStore.LOCK acquired", "hash", chunk.Key.Hex())
+	log.Error("DbStore.LOCK acquired", "hash", chunk.Key.Hex())
 	defer s.lock.Unlock()
 
 	idata, err := s.db.Get(ikey)
-	fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Get done", "hash", chunk.Key.Hex(), "err", err)
+	log.Error("DbStore.db.Get done", "hash", chunk.Key.Hex(), "err", err)
 	if err != nil {
 		s.doPut(chunk, ikey, &index, po)
+		batchC := s.batchC
 		go func() {
 			defer func() {
 				if err := recover(); err != nil {
-					fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put PANIC", "hash", chunk.Key.Hex(), "err", err)
+					log.Error("DbStore.Put PANIC", "hash", chunk.Key.Hex(), "err", err)
 				}
 			}()
-		}()
-		fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put doPut", "hash", chunk.Key.Hex(), "dataIdx", s.dataIdx)
-	} else {
-		log.Trace(fmt.Sprintf("DbStore: chunk already exists, only update access"))
-		decodeIndex(idata, &index)
-		fmt.Fprintln(os.Stderr, time.Now(), "DbStore.Put already found", "hash", chunk.Key.Hex())
-	}
-	batchC := s.batchC
-	go func() {
+
 			<-batchC
 			close(chunk.dbStored)
 		}()
+		log.Error("DbStore.Put doPut", "hash", chunk.Key.Hex(), "dataIdx", s.dataIdx)
+	} else {
+		log.Trace(fmt.Sprintf("DbStore: chunk already exists, only update access"))
+		decodeIndex(idata, &index)
+		close(chunk.dbStored)
+		log.Error("DbStore.Put already found", "hash", chunk.Key.Hex())
+	}
 	index.Access = s.accessCnt
 	s.accessCnt++
 	idata = encodeIndex(&index)
@@ -594,7 +593,7 @@ func (s *DbStore) Put(chunk *Chunk) {
 	case s.batchesC <- struct{}{}:
 	default:
 	}
-	fmt.Fprintln(os.Stderr, time.Now(), "DbStore.db.Put done", "hash", chunk.Key.Hex(), "err", err)
+	log.Error("DbStore.db.Put done", "hash", chunk.Key.Hex(), "err", err)
 }
 
 // force putting into db, does not check access index
@@ -716,9 +715,9 @@ func (s *DbStore) get(key Key) (chunk *Chunk, err error) {
 			hash := hasher.Sum(nil)
 
 			if !bytes.Equal(hash, key) {
-				log.Trace(fmt.Sprintf("Apparent key/hash mismatch. Hash %x, key %v", hash, key[:]))
+				log.Error(fmt.Sprintf("Apparent key/hash mismatch. Hash %x, key %v", hash, key[:]))
 				s.delete(indx.Idx, getIndexKey(key), s.po(key))
-				log.Warn("Invalid Chunk in Database. Please repair with command: 'swarm cleandb'")
+				log.Error("Invalid Chunk in Database. Please repair with command: 'swarm cleandb'")
 			}
 		}
 
@@ -784,15 +783,18 @@ func (s *DbStore) Close() {
 
 // initialises a sync iterator from a syncToken (passed in with the handshake)
 func (s *DbStore) SyncIterator(since uint64, until uint64, po uint8, f func(Key, uint64) bool) error {
-	s.lock.Lock()
-	defer s.lock.Unlock()
+	// probably, the lock is not needed
+	// s.lock.Lock()
+	// defer s.lock.Unlock()
+
 	untilkey := getDataKey(until, po)
 
 	it := s.db.NewIterator()
 	seek := getDataKey(since, po)
 	it.Seek(seek)
 	defer it.Release()
-	for it.Valid() {
+
+	for it.Next() {
 		dbkey := it.Key()
 		if dbkey[0] != keyData || dbkey[1] != byte(po) || bytes.Compare(untilkey, dbkey) < 0 {
 			break
@@ -803,9 +805,8 @@ func (s *DbStore) SyncIterator(since uint64, until uint64, po uint8, f func(Key,
 		if !f(Key(key), binary.BigEndian.Uint64(dbkey[2:])) {
 			break
 		}
-		it.Next()
 	}
-	return nil
+	return it.Error()
 }
 
 func databaseExists(path string) bool {

swarm/storage/localstore.go

@@ -19,9 +19,7 @@ package storage
 import (
 	"encoding/binary"
 	"fmt"
-	"os"
 	"path/filepath"
-	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/swarm/storage/mock"
@@ -98,9 +96,9 @@ func NewTestLocalStoreForAddr(path string, basekey []byte) (*LocalStore, error)
 func (self *LocalStore) Put(chunk *Chunk) {
 	chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
 	self.memStore.Put(chunk)
-	fmt.Fprintln(os.Stderr, time.Now(), "put to memstore", "hash", chunk.Key.Hex())
+	log.Error("put to memstore", "hash", chunk.Key.Hex())
 	self.DbStore.Put(chunk)
-	fmt.Fprintln(os.Stderr, time.Now(), "put to dbstore", "hash", chunk.Key.Hex())
+	log.Error("put to dbstore", "hash", chunk.Key.Hex())
 }
 
 // Get(chunk *Chunk) looks up a chunk in the local stores