swarm/network, swarm/pss: Add context cancels, missing format vars

swarm/network/protocol.go

@@ -207,9 +207,8 @@ func (b *Bzz) RunProtocol(spec *protocols.Spec, run func(*bzzPeer) error) func(*
 // performHandshake implements the negotiation of the bzz handshake
 // shared among swarm subprotocols
 func performHandshake(p *protocols.Peer, handshake *HandshakeMsg) error {
-	ctx, _ := context.WithTimeout(context.Background(), bzzHandshakeTimeout)
+	ctx, cancel := context.WithTimeout(context.Background(), bzzHandshakeTimeout)
-	// defer cancel()
+	defer cancel()
-	// ctx, cancel := context.WithTimeout(context.Background(), bzzHandshakeTimeout)
 	defer close(handshake.done)
 	rsh, err := p.Handshake(ctx, handshake, checkHandshake)
 	if err != nil {

swarm/network/simulations/discovery/discovery_test.go

@@ -137,7 +137,7 @@ func benchmarkDiscovery(b *testing.B, nodes, conns int) {
 	for i := 0; i < b.N; i++ {
 		result, err := discoverySimulation(nodes, conns, adapters.NewSimAdapter(services))
 		if err != nil {
-			b.Fatalf("setting up simulation failed", result)
+			b.Fatalf("setting up simulation failed: %v", err)
 		}
 		if result.Error != nil {
 			b.Logf("simulation failed: %s", result.Error)

swarm/pss/client/client_test.go

@@ -104,7 +104,8 @@ func TestClientHandshake(t *testing.T) {
 	lproto := pss.NewPingProtocol(lpssping)
 	rproto := pss.NewPingProtocol(rpssping)
 
-	ctx, _ := context.WithTimeout(context.Background(), time.Second*10)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
 	err = lpsc.RunProtocol(ctx, lproto)
 	if err != nil {
 		t.Fatal(err)
@@ -231,13 +232,14 @@ func newServices() adapters.Services {
 		"pss": func(ctx *adapters.ServiceContext) (node.Service, error) {
 			cachedir, err := ioutil.TempDir("", "pss-cache")
 			if err != nil {
-				return nil, fmt.Errorf("create pss cache tmpdir failed", "error", err)
+				return nil, fmt.Errorf("create pss cache tmpdir failed: %v", err)
 			}
 			dpa, err := storage.NewLocalDPA(cachedir)
 			if err != nil {
-				return nil, fmt.Errorf("local dpa creation failed", "error", err)
+				return nil, fmt.Errorf("local dpa creation failed: %v", err)
 			}
-			ctxlocal, _ := context.WithTimeout(context.Background(), time.Second)
+			ctxlocal, cancel := context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
 			keys, err := wapi.NewKeyPair(ctxlocal)
 			privkey, err := w.GetPrivateKey(keys)
 			psparams := pss.NewPssParams(privkey)

swarm/pss/handshake.go

@@ -268,7 +268,7 @@ func (self *HandshakeController) handler(msg []byte, p *p2p.Peer, asymmetric boo
 	if !asymmetric {
 		if self.symKeyIndex[symkeyid] != nil {
 			if self.symKeyIndex[symkeyid].count >= self.symKeyIndex[symkeyid].limit {
-				return fmt.Errorf("discarding message using expired key", "symkeyid", symkeyid)
+				return fmt.Errorf("discarding message using expired key: %s", symkeyid)
 			}
 			self.symKeyIndex[symkeyid].count++
 			log.Trace("increment symkey recv use", "symsymkeyid", symkeyid, "count", self.symKeyIndex[symkeyid].count, "limit", self.symKeyIndex[symkeyid].limit, "receiver", common.ToHex(crypto.FromECDSAPub(self.pss.PublicKey())))
@@ -457,7 +457,8 @@ func (self *HandshakeAPI) Handshake(pubkeyid string, topic Topic, sync bool, flu
 		return keys, err
 	}
 	if sync {
-		ctx, _ := context.WithTimeout(context.Background(), self.ctrl.symKeyRequestTimeout)
+		ctx, cancel := context.WithTimeout(context.Background(), self.ctrl.symKeyRequestTimeout)
+		defer cancel()
 		select {
 		case keys = <-hsc:
 			log.Trace("sync handshake response receive", "key", keys)

swarm/pss/protocol.go

@@ -227,7 +227,7 @@ func (self *Protocol) AddPeer(p *p2p.Peer, run func(*p2p.Peer, p2p.MsgReadWriter
 	}
 	go func() {
 		err := run(p, rw)
-		log.Warn(fmt.Sprintf("pss vprotocol quit on addr %v topic %v: %v", topic, err))
+		log.Warn(fmt.Sprintf("pss vprotocol quit on %v topic %v: %v", p, topic, err))
 	}()
 	return rw, nil
 }

swarm/pss/protocol_test.go

@@ -73,11 +73,13 @@ func testProtocol(t *testing.T) {
 	time.Sleep(time.Millisecond * 1000) // replace with hive healthy code
 
 	lmsgC := make(chan APIMsg)
-	lctx, _ := context.WithTimeout(context.Background(), time.Second*10)
+	lctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
 	lsub, err := clients[0].Subscribe(lctx, "pss", lmsgC, "receive", topic)
 	defer lsub.Unsubscribe()
 	rmsgC := make(chan APIMsg)
-	rctx, _ := context.WithTimeout(context.Background(), time.Second*10)
+	rctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
 	rsub, err := clients[1].Subscribe(rctx, "pss", rmsgC, "receive", topic)
 	defer rsub.Unsubscribe()
 

swarm/pss/pss.go

@@ -501,7 +501,7 @@ func (self *Pss) processSym(envelope *whisper.Envelope) (*whisper.ReceivedMessag
 func (self *Pss) processAsym(envelope *whisper.Envelope) (*whisper.ReceivedMessage, string, *PssAddress, error) {
 	recvmsg, err := envelope.OpenAsymmetric(self.privateKey)
 	if err != nil {
-		return nil, "", nil, fmt.Errorf("could not decrypt message: %v", "err", err)
+		return nil, "", nil, fmt.Errorf("could not decrypt message: %v", err)
 	}
 	// check signature (if signed), strip padding
 	if !recvmsg.Validate() {

swarm/pss/pss_test.go

@@ -137,7 +137,8 @@ func TestTopic(t *testing.T) {
 func TestCache(t *testing.T) {
 	var err error
 	to, _ := hex.DecodeString("08090a0b0c0d0e0f1011121314150001020304050607161718191a1b1c1d1e1f")
-	ctx, _ := context.WithTimeout(context.Background(), time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
 	keys, err := wapi.NewKeyPair(ctx)
 	privkey, err := w.GetPrivateKey(keys)
 	if err != nil {
@@ -211,7 +212,8 @@ func TestAddressMatch(t *testing.T) {
 	remoteaddr := []byte("feedbeef")
 	kadparams := network.NewKadParams()
 	kad := network.NewKademlia(localaddr, kadparams)
-	ctx, _ := context.WithTimeout(context.Background(), time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
 	keys, err := wapi.NewKeyPair(ctx)
 	if err != nil {
 		t.Fatalf("Could not generate private key: %v", err)
@@ -255,12 +257,14 @@ func TestAddressMatch(t *testing.T) {
 // set and generate pubkeys and symkeys
 func TestKeys(t *testing.T) {
 	// make our key and init pss with it
-	ctx, _ := context.WithTimeout(context.Background(), time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
 	ourkeys, err := wapi.NewKeyPair(ctx)
 	if err != nil {
 		t.Fatalf("create 'our' key fail")
 	}
-	ctx, _ = context.WithTimeout(context.Background(), time.Second)
+	ctx, cancel = context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
 	theirkeys, err := wapi.NewKeyPair(ctx)
 	if err != nil {
 		t.Fatalf("create 'their' key fail")
@@ -449,12 +453,14 @@ func testSymSend(t *testing.T) {
 	// at this point we've verified that symkeys are saved and match on each peer
 	// now try sending symmetrically encrypted message, both directions
 	lmsgC := make(chan APIMsg)
-	lctx, _ := context.WithTimeout(context.Background(), time.Second*10)
+	lctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
 	lsub, err := clients[0].Subscribe(lctx, "pss", lmsgC, "receive", topic)
 	log.Trace("lsub", "id", lsub)
 	defer lsub.Unsubscribe()
 	rmsgC := make(chan APIMsg)
-	rctx, _ := context.WithTimeout(context.Background(), time.Second*10)
+	rctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
 	rsub, err := clients[1].Subscribe(rctx, "pss", rmsgC, "receive", topic)
 	log.Trace("rsub", "id", rsub)
 	defer rsub.Unsubscribe()
@@ -562,12 +568,14 @@ func testAsymSend(t *testing.T) {
 	time.Sleep(time.Millisecond * 500) // replace with hive healthy code
 
 	lmsgC := make(chan APIMsg)
-	lctx, _ := context.WithTimeout(context.Background(), time.Second*10)
+	lctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
 	lsub, err := clients[0].Subscribe(lctx, "pss", lmsgC, "receive", topic)
 	log.Trace("lsub", "id", lsub)
 	defer lsub.Unsubscribe()
 	rmsgC := make(chan APIMsg)
-	rctx, _ := context.WithTimeout(context.Background(), time.Second*10)
+	rctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
 	rsub, err := clients[1].Subscribe(rctx, "pss", rmsgC, "receive", topic)
 	log.Trace("rsub", "id", rsub)
 	defer rsub.Unsubscribe()
@@ -627,7 +635,6 @@ func worker(id int, jobs <-chan Job, rpcs map[discover.NodeID]*rpc.Client, pubke
 // nodes/msgs/addrbytes/adaptertype
 // if adaptertype is exec uses execadapter, simadapter otherwise
 func TestNetwork(t *testing.T) {
-	t.Skip("skip until proper local benchmark values for stress testing can be determined")
 	t.Run("3/2000/4/sock", testNetwork)
 	t.Run("4/2000/4/sock", testNetwork)
 	t.Run("8/2000/4/sock", testNetwork)
@@ -835,7 +842,8 @@ func benchmarkSymKeySend(b *testing.B) {
 	if err != nil {
 		b.Fatalf("benchmark called with invalid msgsize param '%s': %v", msgsizestring[1], err)
 	}
-	ctx, _ := context.WithTimeout(context.Background(), time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
 	keys, err := wapi.NewKeyPair(ctx)
 	privkey, err := w.GetPrivateKey(keys)
 	ps := newTestPss(privkey, nil, nil)
@@ -878,7 +886,8 @@ func benchmarkAsymKeySend(b *testing.B) {
 	if err != nil {
 		b.Fatalf("benchmark called with invalid msgsize param '%s': %v", msgsizestring[1], err)
 	}
-	ctx, _ := context.WithTimeout(context.Background(), time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
 	keys, err := wapi.NewKeyPair(ctx)
 	privkey, err := w.GetPrivateKey(keys)
 	ps := newTestPss(privkey, nil, nil)
@@ -923,7 +932,8 @@ func benchmarkSymkeyBruteforceChangeaddr(b *testing.B) {
 	}
 	pssmsgs := make([]*PssMsg, 0, keycount)
 	var keyid string
-	ctx, _ := context.WithTimeout(context.Background(), time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
 	keys, err := wapi.NewKeyPair(ctx)
 	privkey, err := w.GetPrivateKey(keys)
 	if cachesize > 0 {
@@ -1005,7 +1015,8 @@ func benchmarkSymkeyBruteforceSameaddr(b *testing.B) {
 		}
 	}
 	addr := make([]PssAddress, keycount)
-	ctx, _ := context.WithTimeout(context.Background(), time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
 	keys, err := wapi.NewKeyPair(ctx)
 	privkey, err := w.GetPrivateKey(keys)
 	if cachesize > 0 {
@@ -1122,17 +1133,18 @@ func newServices() adapters.Services {
 		pssProtocolName: func(ctx *adapters.ServiceContext) (node.Service, error) {
 			cachedir, err := ioutil.TempDir("", "pss-cache")
 			if err != nil {
-				return nil, fmt.Errorf("create pss cache tmpdir failed", "error", err)
+				return nil, fmt.Errorf("create pss cache tmpdir failed: %v", err)
 			}
 			dpa, err := storage.NewLocalDPA(cachedir)
 			if err != nil {
-				return nil, fmt.Errorf("local dpa creation failed", "error", err)
+				return nil, fmt.Errorf("local dpa creation failed: %v", err)
 			}
 
 			// execadapter does not exec init()
 			initTest()
 
-			ctxlocal, _ := context.WithTimeout(context.Background(), time.Second)
+			ctxlocal, cancel := context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
 			keys, err := wapi.NewKeyPair(ctxlocal)
 			privkey, err := w.GetPrivateKey(keys)
 			pssp := NewPssParams(privkey)