Merge pull request #416 from ethersphere/fix-unexpected-eof

swarm: tracing to help fix `unexpected EOF`

cmd/swarm/swarm-smoke/upload_and_sync.go

@@ -40,6 +40,12 @@ func cliUploadAndSync(c *cli.Context) error {
 
 	log.Info("uploaded successfully", "hash", hash, "digest", fmt.Sprintf("%x", fhash))
 
+	if filesize < 10 {
+		time.Sleep(15 * time.Second)
+	} else {
+		time.Sleep(2 * time.Duration(filesize) * time.Second)
+	}
+
 	wg := sync.WaitGroup{}
 	for _, endpoint := range endpoints {
 		endpoint := endpoint
@@ -66,10 +72,10 @@ func cliUploadAndSync(c *cli.Context) error {
 // fetch is getting the requested `hash` from the `endpoint` and compares it with the `original` file
 func fetch(hash string, endpoint string, original []byte, ruid string) error {
 	log.Trace("sleeping", "ruid", ruid)
-	time.Sleep(10 * time.Second)
+	time.Sleep(1 * time.Second)
 
 	log.Trace("http get request", "ruid", ruid, "api", endpoint, "hash", hash)
-	res, err := http.Get(endpoint + "/bzz:/" + hash)
+	res, err := http.Get(endpoint + "/bzz:/" + hash + "/")
 	if err != nil {
 		log.Warn(err.Error(), "ruid", ruid)
 		return err

swarm/api/api.go

@@ -308,7 +308,7 @@ func (self *Api) Put(content, contentType string, toEncrypt bool) (k storage.Key
 // Get uses iterative manifest retrieval and prefix matching
 // to resolve basePath to content using dpa retrieve
 // it returns a section reader, mimeType, status and an error
-func (self *Api) Get(key storage.Key, path string) (reader storage.LazySectionReader, mimeType string, status int, err error) {
+func (self *Api) Get(key storage.Key, path string) (reader *storage.LazyChunkReader, mimeType string, status int, err error) {
 	log.Debug("api.get", "key", key, "path", path)
 	apiGetCount.Inc(1)
 	trie, err := loadManifest(self.dpa, key, nil)

swarm/network/stream/syncer.go

@@ -130,7 +130,7 @@ func (s *SwarmSyncerServer) SetNextBatch(from, to uint64) ([]byte, uint64, uint6
 		ticker.Stop()
 	}
 
-	log.Debug("Swarm syncer offer batch", "po", s.po, "len", i, "from", from, "to", to, "current store count", s.db.CurrentBucketStorageIndex(s.po))
+	log.Trace("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, nil, nil
 }
 

swarm/storage/chunker.go

@@ -131,7 +131,7 @@ type TreeChunker struct {
 	The chunks are not meant to be validated by the chunker when joining. This
 	is because it is left to the DPA to decide which sources are trusted.
 */
-func TreeJoin(key Key, getter Getter, depth int) LazySectionReader {
+func TreeJoin(key Key, getter Getter, depth int) *LazyChunkReader {
 	return NewTreeJoiner(NewJoinerParams(key, getter, depth, DefaultChunkSize)).Join()
 }
 
@@ -392,8 +392,7 @@ type LazyChunkReader struct {
 	getter    Getter
 }
 
-// implements the Joiner interface
+func (self *TreeChunker) Join() *LazyChunkReader {
-func (self *TreeChunker) Join() LazySectionReader {
 	return &LazyChunkReader{
 		key:       self.key,
 		chunkSize: self.chunkSize,
@@ -436,6 +435,7 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
 	quitC := make(chan bool)
 	size, err := self.Size(quitC)
 	if err != nil {
+		log.Error("lazychunkreader.readat.size", "size", size, "err", err)
 		return 0, err
 	}
 
@@ -464,6 +464,7 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
 
 	err = <-errC
 	if err != nil {
+		log.Error("lazychunkreader.readat.errc", "err", err)
 		close(quitC)
 		return 0, err
 	}
@@ -517,8 +518,9 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr
 			childKey := chunkData[8+j*self.hashSize : 8+(j+1)*self.hashSize]
 			chunkData, err := self.getter.Get(Reference(childKey))
 			if err != nil {
+				log.Error("lazychunkreader.join", "key", fmt.Sprintf("%x", childKey), "err", err)
 				select {
-				case errC <- fmt.Errorf("chunk %v-%v not found", off, off+treeSize):
+				case errC <- fmt.Errorf("chunk %v-%v not found; key: %s", off, off+treeSize, fmt.Sprintf("%x", childKey)):
 				case <-quitC:
 				}
 				return
@@ -535,6 +537,9 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr
 func (self *LazyChunkReader) Read(b []byte) (read int, err error) {
 	log.Debug("lazychunkreader.read", "key", self.key)
 	read, err = self.ReadAt(b, self.off)
+	if err != nil && err != io.EOF {
+		log.Error("lazychunkreader.readat", "read", read, "err", err)
+	}
 
 	self.off += int64(read)
 	return

swarm/storage/dpa.go

@@ -84,7 +84,7 @@ func NewDPA(store ChunkStore, params *DPAParams) *DPA {
 // Chunk retrieval blocks on netStore requests with a timeout so reader will
 // report error if retrieval of chunks within requested range time out.
 // It returns a reader with the chunk data and whether the content was encrypted
-func (self *DPA) Retrieve(key Key) (reader LazySectionReader, isEncrypted bool) {
+func (self *DPA) Retrieve(key Key) (reader *LazyChunkReader, isEncrypted bool) {
 	isEncrypted = len(key) > self.hashFunc().Size()
 	getter := NewHasherStore(self.ChunkStore, self.hashFunc, isEncrypted)
 	reader = TreeJoin(key, getter, 0)