swarm/network: disallow historical retrieval requests (#17936)

swarm/network/stream/delivery_test.go

@@ -416,7 +416,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 				return fmt.Errorf("No registry")
 			}
 			registry := item.(*Registry)
-			err = registry.Subscribe(sid, NewStream(swarmChunkServerStreamName, "", true), NewRange(0, 0), Top)
+			err = registry.Subscribe(sid, NewStream(swarmChunkServerStreamName, "", true), nil, Top)
 			if err != nil {
 				return err
 			}

swarm/network/stream/messages.go

@@ -158,6 +158,7 @@ type SubscribeErrorMsg struct {
 }
 
 func (p *Peer) handleSubscribeErrorMsg(req *SubscribeErrorMsg) (err error) {
+	//TODO the error should be channeled to whoever calls the subscribe
 	return fmt.Errorf("subscribe to peer %s: %v", p.ID(), req.Error)
 }
 

swarm/network/stream/snapshot_retrieval_test.go

@@ -218,7 +218,7 @@ func runFileRetrievalTest(nodeCount int) error {
 					reader, _ := fileStore.Retrieve(context.TODO(), hash)
 					//check that we can read the file size and that it corresponds to the generated file size
 					if s, err := reader.Size(ctx, nil); err != nil || s != int64(len(randomFiles[i])) {
-						log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id)
+						log.Debug("Retrieve error", "err", err, "hash", hash, "nodeId", id)
 						time.Sleep(500 * time.Millisecond)
 						continue REPEAT
 					}
@@ -309,7 +309,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
 					reader, _ := fileStore.Retrieve(context.TODO(), hash)
 					//check that we can read the chunk size and that it corresponds to the generated chunk size
 					if s, err := reader.Size(ctx, nil); err != nil || s != int64(chunkSize) {
-						log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id, "size", s)
+						log.Debug("Retrieve error", "err", err, "hash", hash, "nodeId", id, "size", s)
 						time.Sleep(500 * time.Millisecond)
 						continue REPEAT
 					}

swarm/network/stream/snapshot_sync_test.go

@@ -310,7 +310,7 @@ func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulatio
 						_, err = lstore.Get(ctx, chunk)
 					}
 					if err != nil {
-						log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
+						log.Debug(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
 						// Do not get crazy with logging the warn message
 						time.Sleep(500 * time.Millisecond)
 						continue REPEAT
@@ -514,7 +514,7 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int)
 						_, err = lstore.Get(ctx, chunk)
 					}
 					if err != nil {
-						log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
+						log.Debug(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
 						// Do not get crazy with logging the warn message
 						time.Sleep(500 * time.Millisecond)
 						continue REPEAT

swarm/network/stream/stream.go

@@ -18,6 +18,7 @@ package stream
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
 	"sync"
@@ -96,7 +97,10 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy
 	delivery.getPeer = streamer.getPeer
 
 	if options.DoServeRetrieve {
-		streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, _ bool) (Server, error) {
+		streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, live bool) (Server, error) {
+			if !live {
+				return nil, errors.New("only live retrieval requests supported")
+			}
 			return NewSwarmChunkServer(delivery.chunkStore), nil
 		})
 	}
@@ -279,7 +283,6 @@ func (r *Registry) Subscribe(peerId enode.ID, s Stream, h *Range, priority uint8
 	if err != nil {
 		return err
 	}
-
 	if s.Live && h != nil {
 		if err := peer.setClientParams(
 			getHistoryStream(s),