Send an expiration signal when envelope wasn't delivered to any peer

geth/node/node.go

@@ -213,7 +213,7 @@ func activateShhService(stack *node.Node, config *params.NodeConfig) (err error)
 			return nil, err
 		}
 
-		svc := shhext.New(whisper, shhext.SendEnvelopeSentSignal)
+		svc := shhext.New(whisper, shhext.EnvelopeSignalHandler{})
 		return svc, nil
 	})
 	if err != nil {

geth/signal/signals.go

@@ -33,6 +33,10 @@ const (
 
 	// EventEnvelopeSent is triggered when envelope was sent atleast to a one peer.
 	EventEnvelopeSent = "envelope.sent"
+
+	// EventEnvelopeExpired is triggered when envelop was dropped by a whisper without being sent
+	// to any peer
+	EventEnvelopeExpired = "envelope.expired"
 )
 
 // Envelope is a general signal sent upward from node to RN app

shhext/README.md

@@ -15,7 +15,7 @@ Accepts same input as shh_post (see https://github.com/ethereum/wiki/wiki/JSON-R
 Signals
 -------
 
-Sends following event once per envelope.
+Sends sent signal once per envelope.
 
 ```json
 {
@@ -25,3 +25,15 @@ Sends following event once per envelope.
   }
 }
 ```
+
+Sends expired signal if envelope dropped from whisper local queue before it was
+sent to any peer on the network.
+
+```json
+{
+  "type": "envelope.expired",
+  "event": {
+    "hash": "0x754f4c12dccb14886f791abfeb77ffb86330d03d5a4ba6f37a8c21281988b69e"
+  }
+}
+```

shhext/service.go

@@ -21,8 +21,11 @@ const (
 	EnvelopeSent
 )
 
-// ConfirmationHandler used as a callback for confirming that envelopes were sent.
+// EnvelopeEventsHandler used for two different event types.
-type ConfirmationHandler func(common.Hash)
+type EnvelopeEventsHandler interface {
+	EnvelopeSent(common.Hash)
+	EnvelopeExpired(common.Hash)
+}
 
 // Service is a service that provides some additional Whisper API.
 type Service struct {
@@ -34,7 +37,7 @@ type Service struct {
 var _ node.Service = (*Service)(nil)
 
 // New returns a new Service.
-func New(w *whisper.Whisper, handler ConfirmationHandler) *Service {
+func New(w *whisper.Whisper, handler EnvelopeEventsHandler) *Service {
 	track := &tracker{
 		w:       w,
 		handler: handler,
@@ -81,7 +84,7 @@ func (s *Service) Stop() error {
 // and calling specified handler.
 type tracker struct {
 	w       *whisper.Whisper
-	handler ConfirmationHandler
+	handler EnvelopeEventsHandler
 
 	mu    sync.Mutex
 	cache map[common.Hash]EnvelopeState
@@ -141,15 +144,21 @@ func (t *tracker) handleEvent(event whisper.EnvelopeEvent) {
 		if !ok || state == EnvelopeSent {
 			return
 		}
+		log.Debug("envelope is sent", "hash", event.Hash, "peer", event.Peer)
+		t.cache[event.Hash] = EnvelopeSent
 		if t.handler != nil {
-			log.Debug("envelope is sent", "hash", event.Hash, "peer", event.Peer)
+			t.handler.EnvelopeSent(event.Hash)
-			t.handler(event.Hash)
-			t.cache[event.Hash] = EnvelopeSent
 		}
 	case whisper.EventEnvelopeExpired:
-		if _, ok := t.cache[event.Hash]; ok {
+		if state, ok := t.cache[event.Hash]; ok {
-			log.Debug("envelope expired", "hash", event.Hash)
+			log.Debug("envelope expired", "hash", event.Hash, "state", state)
 			delete(t.cache, event.Hash)
+			if state == EnvelopeSent {
+				return
+			}
+			if t.handler != nil {
+				t.handler.EnvelopeExpired(event.Hash)
+			}
 		}
 	}
 }

shhext/service_test.go

@@ -12,6 +12,26 @@ import (
 	"github.com/stretchr/testify/suite"
 )
 
+func newHandlerMock(buf int) handlerMock {
+	return handlerMock{
+		confirmations: make(chan common.Hash, buf),
+		expirations:   make(chan common.Hash, buf),
+	}
+}
+
+type handlerMock struct {
+	confirmations chan common.Hash
+	expirations   chan common.Hash
+}
+
+func (t handlerMock) EnvelopeSent(hash common.Hash) {
+	t.confirmations <- hash
+}
+
+func (t handlerMock) EnvelopeExpired(hash common.Hash) {
+	t.expirations <- hash
+}
+
 func TestShhExtSuite(t *testing.T) {
 	suite.Run(t, new(ShhExtSuite))
 }
@@ -28,15 +48,14 @@ func (s *ShhExtSuite) SetupTest() {
 	s.nodes = make([]*node.Node, 2)
 	s.services = make([]*Service, 2)
 	s.whisper = make([]*whisper.Whisper, 2)
-	port := 21313
 	for i := range s.nodes {
 		i := i // bind i to be usable in service constructors
 		cfg := &node.Config{
 			Name: fmt.Sprintf("node-%d", i),
 			P2P: p2p.Config{
 				NoDiscovery: true,
-				MaxPeers:    20,
+				MaxPeers:    1,
-				ListenAddr:  fmt.Sprintf(":%d", port+i),
+				ListenAddr:  ":0",
 			},
 		}
 		stack, err := node.New(cfg)
@@ -52,15 +71,13 @@ func (s *ShhExtSuite) SetupTest() {
 		s.Require().NoError(stack.Start())
 		s.nodes[i] = stack
 	}
-	s.nodes[0].Server().AddPeer(s.nodes[1].Server().Self())
+	s.services[0].tracker.handler = newHandlerMock(1)
 }
 
 func (s *ShhExtSuite) TestPostMessageWithConfirmation() {
-	confirmations := make(chan common.Hash, 1)
+	mock := newHandlerMock(1)
-	confirmationsHandler := func(hash common.Hash) {
+	s.services[0].tracker.handler = mock
-		confirmations <- hash
+	s.nodes[0].Server().AddPeer(s.nodes[1].Server().Self())
-	}
-	s.services[0].tracker.handler = confirmationsHandler
 	symID, err := s.whisper[0].GenerateSymKey()
 	s.NoError(err)
 	client, err := s.nodes[0].Attach()
@@ -75,13 +92,40 @@ func (s *ShhExtSuite) TestPostMessageWithConfirmation() {
 	}))
 	s.NoError(err)
 	select {
-	case confirmed := <-confirmations:
+	case confirmed := <-mock.confirmations:
 		s.Equal(hash, confirmed)
 	case <-time.After(time.Second):
 		s.Fail("timed out while waiting for confirmation")
 	}
 }
 
+func (s *ShhExtSuite) TestWaitMessageExpired() {
+	mock := newHandlerMock(1)
+	s.services[0].tracker.handler = mock
+	symID, err := s.whisper[0].GenerateSymKey()
+	s.NoError(err)
+	client, err := s.nodes[0].Attach()
+	s.NoError(err)
+	var hash common.Hash
+	s.NoError(client.Call(&hash, "shhext_post", whisper.NewMessage{
+		SymKeyID:  symID,
+		PowTarget: whisper.DefaultMinimumPoW,
+		PowTime:   200,
+		TTL:       1,
+		Topic:     whisper.TopicType{0x01, 0x01, 0x01, 0x01},
+		Payload:   []byte("hello"),
+	}))
+	s.NoError(err)
+	select {
+	case expired := <-mock.expirations:
+		s.Equal(hash, expired)
+	case confirmed := <-mock.confirmations:
+		s.Fail("unexpected confirmation for hash", confirmed)
+	case <-time.After(2 * time.Second):
+		s.Fail("timed out while waiting for confirmation")
+	}
+}
+
 func (s *ShhExtSuite) TearDown() {
 	for _, n := range s.nodes {
 		s.NoError(n.Stop())
@@ -104,8 +148,7 @@ type TrackerSuite struct {
 
 func (s *TrackerSuite) SetupTest() {
 	s.tracker = &tracker{
-		handler: func(common.Hash) {},
+		cache: map[common.Hash]EnvelopeState{},
-		cache:   map[common.Hash]EnvelopeState{},
 	}
 }
 

shhext/signal.go

@@ -5,17 +5,26 @@ import (
 	"github.com/status-im/status-go/geth/signal"
 )
 
-// EnvelopeSentSignal includes hash of the sent envelope.
+// EnvelopeSignal includes hash of the envelope.
-type EnvelopeSentSignal struct {
+type EnvelopeSignal struct {
 	Hash common.Hash `json:"hash"`
 }
 
-// SendEnvelopeSentSignal sends an envelope.sent signal with hash of the envelope.
+// EnvelopeSignalHandler sends signals when envelope is sent or expired.
-func SendEnvelopeSentSignal(hash common.Hash) {
+type EnvelopeSignalHandler struct{}
+
+// EnvelopeSent triggered when envelope delivered atleast to 1 peer.
+func (h EnvelopeSignalHandler) EnvelopeSent(hash common.Hash) {
 	signal.Send(signal.Envelope{
-		Type: signal.EventEnvelopeSent,
+		Type:  signal.EventEnvelopeSent,
-		Event: EnvelopeSentSignal{
+		Event: EnvelopeSignal{Hash: hash},
-			Hash: hash,
+	})
-		},
+}
+
+// EnvelopeExpired triggered when envelope is expired but wasn't delivered to any peer.
+func (h EnvelopeSignalHandler) EnvelopeExpired(hash common.Hash) {
+	signal.Send(signal.Envelope{
+		Type:  signal.EventEnvelopeExpired,
+		Event: EnvelopeSignal{Hash: hash},
 	})
 }

t/e2e/whisper/whisper_ext_test.go

@@ -17,17 +17,17 @@ import (
 	"github.com/stretchr/testify/suite"
 )
 
-func TestWhisperExtentionSuite(t *testing.T) {
+func TestWhisperExtensionSuite(t *testing.T) {
-	suite.Run(t, new(WhisperExtentionSuite))
+	suite.Run(t, new(WhisperExtensionSuite))
 }
 
-type WhisperExtentionSuite struct {
+type WhisperExtensionSuite struct {
 	suite.Suite
 
 	nodes []*node.StatusNode
 }
 
-func (s *WhisperExtentionSuite) SetupTest() {
+func (s *WhisperExtensionSuite) SetupTest() {
 	s.nodes = make([]*node.StatusNode, 2)
 	for i := range s.nodes {
 		dir, err := ioutil.TempDir("", "test-shhext-")
@@ -40,14 +40,14 @@ func (s *WhisperExtentionSuite) SetupTest() {
 		s.nodes[i] = node.New()
 		s.Require().NoError(s.nodes[i].Start(cfg))
 	}
+}
+
+func (s *WhisperExtensionSuite) TestSentSignal() {
 	node1, err := s.nodes[0].GethNode()
 	s.NoError(err)
 	node2, err := s.nodes[1].GethNode()
 	s.NoError(err)
 	node1.Server().AddPeer(node2.Server().Self())
-}
-
-func (s *WhisperExtentionSuite) TestRecievedSignal() {
 	confirmed := make(chan common.Hash, 1)
 	signal.SetDefaultNodeNotificationHandler(func(rawSignal string) {
 		var sg struct {
@@ -57,7 +57,7 @@ func (s *WhisperExtentionSuite) TestRecievedSignal() {
 		s.NoError(json.Unmarshal([]byte(rawSignal), &sg))
 
 		if sg.Type == signal.EventEnvelopeSent {
-			var event shhext.EnvelopeSentSignal
+			var event shhext.EnvelopeSignal
 			s.NoError(json.Unmarshal(sg.Event, &event))
 			confirmed <- event.Hash
 		}
@@ -84,7 +84,46 @@ func (s *WhisperExtentionSuite) TestRecievedSignal() {
 	}
 }
 
-func (s *WhisperExtentionSuite) TearDown() {
+func (s *WhisperExtensionSuite) TestExpiredSignal() {
+	expired := make(chan common.Hash, 1)
+	signal.SetDefaultNodeNotificationHandler(func(rawSignal string) {
+		var sg struct {
+			Type  string
+			Event json.RawMessage
+		}
+		fmt.Println(string(rawSignal))
+		s.NoError(json.Unmarshal([]byte(rawSignal), &sg))
+
+		if sg.Type == signal.EventEnvelopeExpired {
+			var event shhext.EnvelopeSignal
+			s.NoError(json.Unmarshal(sg.Event, &event))
+			expired <- event.Hash
+		}
+	})
+	client := s.nodes[0].RPCClient()
+	s.NotNil(client)
+	var symID string
+	s.NoError(client.Call(&symID, "shh_newSymKey"))
+	msg := whisper.NewMessage{
+		SymKeyID:  symID,
+		PowTarget: whisper.DefaultMinimumPoW,
+		PowTime:   200,
+		TTL:       1,
+		Topic:     whisper.TopicType{0x01, 0x01, 0x01, 0x01},
+		Payload:   []byte("hello"),
+	}
+	var hash common.Hash
+	s.NoError(client.Call(&hash, "shhext_post", msg))
+	s.NotEqual(common.Hash{}, hash)
+	select {
+	case exp := <-expired:
+		s.Equal(hash, exp)
+	case <-time.After(3 * time.Second):
+		s.Fail("timed out while waiting for expiration")
+	}
+}
+
+func (s *WhisperExtensionSuite) TearDown() {
 	for _, n := range s.nodes {
 		cfg, err := n.Config()
 		s.NoError(err)