swarm/network/stream: Different chunk delivery msg types for syncing and retrieval

swarm/network/stream/delivery.go

@@ -197,6 +197,7 @@ func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *
 	return nil
 }
 
+//Chunk delivery always uses the same message type....
 type ChunkDeliveryMsg struct {
 	Addr    storage.Address
 	SData   []byte // the stored chunk Data (incl size)
@@ -204,6 +205,15 @@ type ChunkDeliveryMsg struct {
 	Syncing bool   // if true, this is a delivery for syncing (no SWAP accounting needed)
 }
 
+//...but swap accounting needs to disambiguate if it is a delivery for syncing or for retrieval
+//as it decides based on message type if it needs to account for this message or not
+
+//defines a chunk delivery for retrieval (with accounting)
+type ChunkDeliveryMsgRetrieval ChunkDeliveryMsg
+
+//defines a chunk delivery for syncing (without accounting)
+type ChunkDeliveryMsgSyncing ChunkDeliveryMsg
+
 // TODO: Fix context SNAFU
 func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req *ChunkDeliveryMsg) error {
 	var osp opentracing.Span

swarm/network/stream/peer.go

@@ -130,6 +130,7 @@ func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer {
 }
 
 // Deliver sends a storeRequestMsg protocol message to the peer
+// Depending on the `syncing` parameter we send different message types
 func (p *Peer) Deliver(ctx context.Context, chunk storage.Chunk, priority uint8, syncing bool) error {
 	var sp opentracing.Span
 	ctx, sp = spancontext.StartSpan(
@@ -137,10 +138,23 @@ func (p *Peer) Deliver(ctx context.Context, chunk storage.Chunk, priority uint8,
 		"send.chunk.delivery")
 	defer sp.Finish()
 
-	msg := &ChunkDeliveryMsg{
+	var msg interface{}
-		Addr:    chunk.Address(),
+
-		SData:   chunk.Data(),
+	//we send different types of messages if delivery is for syncing or retrievals,
-		Syncing: syncing,
+	//even if handling and content of the message are the same,
+	//because swap accounting decides which messages need accounting based on the message type
+	if syncing {
+		msg = &ChunkDeliveryMsgSyncing{
+			Addr:    chunk.Address(),
+			SData:   chunk.Data(),
+			Syncing: syncing,
+		}
+	} else {
+		msg = &ChunkDeliveryMsgRetrieval{
+			Addr:    chunk.Address(),
+			SData:   chunk.Data(),
+			Syncing: syncing,
+		}
 	}
 	return p.SendPriority(ctx, msg, priority)
 }

swarm/network/stream/stream.go

@@ -482,8 +482,13 @@ func (p *Peer) HandleMsg(ctx context.Context, msg interface{}) error {
 	case *WantedHashesMsg:
 		return p.handleWantedHashesMsg(ctx, msg)
 
-	case *ChunkDeliveryMsg:
+	case *ChunkDeliveryMsgRetrieval:
-		return p.streamer.delivery.handleChunkDeliveryMsg(ctx, p, msg)
+		//handling chunk delivery is the same for retrieval and syncing, so let's cast the msg
+		return p.streamer.delivery.handleChunkDeliveryMsg(ctx, p, ((*ChunkDeliveryMsg)(msg)))
+
+	case *ChunkDeliveryMsgSyncing:
+		//handling chunk delivery is the same for retrieval and syncing, so let's cast the msg
+		return p.streamer.delivery.handleChunkDeliveryMsg(ctx, p, ((*ChunkDeliveryMsg)(msg)))
 
 	case *RetrieveRequestMsg:
 		return p.streamer.delivery.handleRetrieveRequestMsg(ctx, p, msg)
@@ -683,10 +688,11 @@ var Spec = &protocols.Spec{
 		TakeoverProofMsg{},
 		SubscribeMsg{},
 		RetrieveRequestMsg{},
-		ChunkDeliveryMsg{},
+		ChunkDeliveryMsgRetrieval{},
 		SubscribeErrorMsg{},
 		RequestSubscriptionMsg{},
 		QuitMsg{},
+		ChunkDeliveryMsgSyncing{},
 	},
 }