Events, panics & deltas

This commit has a few changes & bug fixes:

* Fixes a panic when using `WithTLS` with a `nil` TLS config
* Adds the `Events` method which returns a read-only channel with state
  updates for nodes. Each event contains a field denoting the type of
  change and the related peer's data.
* Nodes no longer update their own delta every time they send it to a
  peer. With this, the events stream was super noisy with a lot of
  useless events. The caveat of this is that a node that is marked as
  gone may need to rejoin or perform a metadata update.
* Nodes now share state with up to 3 peers at once to speed up
  convergance.

Signed-off-by: David Bond <[email protected]>

Author: davidsbond

README.md

@@ -82,6 +82,14 @@ Below is a very concise example of starting a whisper node in-code:
 // See the package documentation for all available configuration options.
 node := whisper.New(id)
 
+// This is how you react to changes within the gossip network, by handling individual events. You'll want to do this
+// in its own goroutine. Events MUST be handled otherwise the node will eventually block on writing to the channel.
+for evt := range node.Events() {
+    switch evt.Type {
+        // ...
+    }
+}
+
 // This blocks until the given context is cancelled or a fatal error occurs, use it in a separate goroutine or
 // an error group
 node.Run(ctx)
@@ -105,17 +113,17 @@ within the network.
 
 ```go
 // The PeerStore interface describes types that can persist peer data.
-PeerStore interface {
-// FindPeer should return the peer.Peer whose identifier matches the one provided. It should return
-// store.ErrPeerNotFound if a matching peer does not exist.
-FindPeer(ctx context.Context, id uint64) (peer.Peer, error)
-// SavePeer should persist the given peer.Peer.
-SavePeer(ctx context.Context, peer peer.Peer) error
-// ListPeers should return all peers within the store.
-ListPeers(ctx context.Context) ([]peer.Peer, error)
-// RemovePeer should remove a peer from the store. It should return store.ErrPeerNotFound if a matching
-// peer does not exist.
-RemovePeer(ctx context.Context, id uint64) error
+type PeerStore interface {
+    // FindPeer should return the peer.Peer whose identifier matches the one provided. It should return
+    // store.ErrPeerNotFound if a matching peer does not exist.
+    FindPeer(ctx context.Context, id uint64) (peer.Peer, error)
+    // SavePeer should persist the given peer.Peer.
+    SavePeer(ctx context.Context, peer peer.Peer) error
+    // ListPeers should return all peers within the store.
+    ListPeers(ctx context.Context) ([]peer.Peer, error)
+    // RemovePeer should remove a peer from the store. It should return store.ErrPeerNotFound if a matching
+    // peer does not exist.
+    RemovePeer(ctx context.Context, id uint64) error
 }
 ```
 

cmd/whisper/start/start.go

@@ -9,9 +9,11 @@ import (
 	"strconv"
 
 	"github.com/spf13/cobra"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/davidsbond/whisper"
 	"github.com/davidsbond/whisper/internal/tlsutil"
+	"github.com/davidsbond/whisper/pkg/event"
 )
 
 func Command() *cobra.Command {
@@ -64,17 +66,45 @@ func Command() *cobra.Command {
 				}
 			}
 
+			logger := slog.Default().With("local_id", id)
+
 			node := whisper.New(id,
 				whisper.WithPort(port),
 				whisper.WithAddress(address),
 				whisper.WithJoinAddress(joinAddress),
-				whisper.WithLogger(slog.Default().With("local_id", id)),
+				whisper.WithLogger(logger),
 				whisper.WithCurve(curve),
 				whisper.WithKey(key),
 				whisper.WithTLS(tlsConfig),
 			)
 
-			return node.Run(cmd.Context())
+			group, ctx := errgroup.WithContext(cmd.Context())
+			group.Go(func() error {
+				return node.Run(ctx)
+			})
+
+			group.Go(func() error {
+				for evt := range node.Events() {
+					switch evt.Type {
+					case event.TypeDiscovered:
+						logger.With("peer", evt.Peer.ID).InfoContext(ctx, "peer discovered")
+					case event.TypeUpdated:
+						logger.With("peer", evt.Peer.ID).InfoContext(ctx, "peer updated")
+					case event.TypeRemoved:
+						logger.With("peer", evt.Peer.ID).WarnContext(ctx, "peer removed")
+					case event.TypeLeft:
+						logger.With("peer", evt.Peer.ID).WarnContext(ctx, "peer left")
+					case event.TypeGone:
+						logger.With("peer", evt.Peer.ID).ErrorContext(ctx, "peer gone")
+					default:
+						continue
+					}
+				}
+
+				return nil
+			})
+
+			return group.Wait()
 		},
 	}
 

internal/service/service.go

@@ -14,10 +14,10 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
-	"google.golang.org/protobuf/types/known/anypb"
 
 	whispersvcv1 "github.com/davidsbond/whisper/internal/generated/proto/whisper/service/v1"
 	whisperv1 "github.com/davidsbond/whisper/internal/generated/proto/whisper/v1"
+	"github.com/davidsbond/whisper/pkg/event"
 	"github.com/davidsbond/whisper/pkg/peer"
 	"github.com/davidsbond/whisper/pkg/store"
 )
@@ -33,6 +33,7 @@ type (
 		curve  ecdh.Curve
 		logger *slog.Logger
 		tls    *tls.Config
+		events chan<- event.Event
 	}
 
 	// The PeerStore interface describes types that persist the current state of all peers within the gossip network.
@@ -48,13 +49,14 @@ type (
 )
 
 // New returns a new instance of the Service type that will persist peer data using the provided PeerStore implementation.
-func New(id uint64, peers PeerStore, curve ecdh.Curve, logger *slog.Logger, tls *tls.Config) *Service {
+func New(id uint64, peers PeerStore, curve ecdh.Curve, logger *slog.Logger, tls *tls.Config, events chan<- event.Event) *Service {
 	return &Service{
 		id:     id,
 		peers:  peers,
 		curve:  curve,
 		logger: logger,
 		tls:    tls,
+		events: events,
 	}
 }
 
@@ -103,23 +105,23 @@ func (svc *Service) Join(ctx context.Context, r *whispersvcv1.JoinRequest) (*whi
 
 	response := &whispersvcv1.JoinResponse{Peers: make([]*whisperv1.Peer, len(peers))}
 	for i, p := range peers {
-		response.Peers[i] = &whisperv1.Peer{
-			Id:        p.ID,
-			Address:   p.Address,
-			PublicKey: p.PublicKey.Bytes(),
-			Delta:     p.Delta,
-			Status:    whisperv1.PeerStatus(p.Status),
+		response.Peers[i], err = peer.ToProto(p)
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, "failed to parse peer %d: %v", p.ID, err)
 		}
+	}
 
-		if p.Metadata != nil {
-			response.Peers[i].Metadata, err = anypb.New(p.Metadata)
-			if err != nil {
-				return nil, status.Errorf(codes.Internal, "failed to marshal metadata for peer %d: %v", p.ID, err)
-			}
-		}
+	eventType := event.TypeDiscovered
+	if !existing.IsEmpty() {
+		eventType = event.TypeUpdated
 	}
 
-	return response, nil
+	select {
+	case <-ctx.Done():
+		return nil, status.FromContextError(ctx.Err()).Err()
+	case svc.events <- event.Event{Type: eventType, Peer: p}:
+		return response, nil
+	}
 }
 
 func (svc *Service) validateJoinRequest(r *whispersvcv1.JoinRequest) error {
@@ -170,7 +172,13 @@ func (svc *Service) Leave(ctx context.Context, r *whispersvcv1.LeaveRequest) (*w
 
 	svc.logger.With("peer", r.GetId()).DebugContext(ctx, "peer is leaving gossip network")
 
-	return &whispersvcv1.LeaveResponse{}, nil
+	// We have to publish an event here, otherwise the local peer will never hear about peers leaving.
+	select {
+	case <-ctx.Done():
+		return nil, status.FromContextError(ctx.Err()).Err()
+	case svc.events <- event.Event{Type: event.TypeLeft, Peer: p}:
+		return &whispersvcv1.LeaveResponse{}, nil
+	}
 }
 
 // Status handles an inbound gRPC request querying this peer's current view of the gossip network.
@@ -185,19 +193,9 @@ func (svc *Service) Status(ctx context.Context, _ *whispersvcv1.StatusRequest) (
 	}
 
 	for _, p := range peers {
-		record := &whisperv1.Peer{
-			Id:        p.ID,
-			Address:   p.Address,
-			PublicKey: p.PublicKey.Bytes(),
-			Delta:     p.Delta,
-			Status:    whisperv1.PeerStatus(p.Status),
-		}
-
-		if p.Metadata != nil {
-			record.Metadata, err = anypb.New(p.Metadata)
-			if err != nil {
-				return nil, status.Errorf(codes.Internal, "failed to marshal metadata for peer %d: %v", p.ID, err)
-			}
+		record, err := peer.ToProto(p)
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, "failed to parse peer %d: %v", p.ID, err)
 		}
 
 		if p.ID == svc.id {

internal/service/service_test.go

@@ -1,6 +1,7 @@
 package service_test
 
 import (
+	"context"
 	"crypto/ecdh"
 	"crypto/rand"
 	"log/slog"
@@ -18,6 +19,7 @@ import (
 	whispersvcv1 "github.com/davidsbond/whisper/internal/generated/proto/whisper/service/v1"
 	whisperv1 "github.com/davidsbond/whisper/internal/generated/proto/whisper/v1"
 	"github.com/davidsbond/whisper/internal/service"
+	"github.com/davidsbond/whisper/pkg/event"
 	"github.com/davidsbond/whisper/pkg/peer"
 	"github.com/davidsbond/whisper/pkg/store"
 )
@@ -178,7 +180,9 @@ func TestService_Join(t *testing.T) {
 				tc.Setup(s)
 			}
 
-			response, err := service.New(tc.ID, s, curve, logger, nil).Join(t.Context(), tc.Request)
+			events := make(chan event.Event, 1)
+
+			response, err := service.New(tc.ID, s, curve, logger, nil, events).Join(t.Context(), tc.Request)
 			if tc.ExpectsError {
 				require.Error(t, err)
 				assert.Nil(t, response)
@@ -197,6 +201,16 @@ func TestService_Join(t *testing.T) {
 				assert.EqualValues(t, expected.GetMetadata(), actual.GetMetadata())
 				assert.EqualValues(t, expected.GetPublicKey(), actual.GetPublicKey())
 			}
+
+			ctx, cancel := context.WithTimeout(t.Context(), time.Second)
+			defer cancel()
+
+			select {
+			case evt := <-events:
+				assert.EqualValues(t, tc.Request.GetPeer().GetId(), evt.Peer.ID)
+			case <-ctx.Done():
+				assert.Fail(t, "timed out waiting for event")
+			}
 		})
 	}
 }

pkg/event/event.go

@@ -0,0 +1,49 @@
+package event
+
+import (
+	"github.com/davidsbond/whisper/pkg/peer"
+)
+
+type (
+	// The Event type represents a single change in peer state.
+	Event struct {
+		// The type of event that has occurred.
+		Type Type
+		// The peer that has changed.
+		Peer peer.Peer
+	}
+
+	// Type is used to denote the kind of state change that has occurred to a single peer.
+	Type int
+)
+
+const (
+	TypeUnspecified Type = iota
+	// TypeDiscovered denotes that a Peer has been newly discovered by the local peer.
+	TypeDiscovered
+	// TypeUpdated denotes that a Peer's delta has changed, possibly due to a metadata update.
+	TypeUpdated
+	// TypeLeft denotes that a Peer has left the gossip network.
+	TypeLeft
+	// TypeGone denotes that a Peer has failed or is not accessible from other peers within the gossip network.
+	TypeGone
+	// TypeRemoved denotes that a Peer has been completely removed from storage.
+	TypeRemoved
+)
+
+func (e Type) String() string {
+	switch e {
+	case TypeDiscovered:
+		return "DISCOVERED"
+	case TypeUpdated:
+		return "UPDATED"
+	case TypeLeft:
+		return "LEFT"
+	case TypeGone:
+		return "GONE"
+	case TypeRemoved:
+		return "REMOVED"
+	default:
+		return "UNSPECIFIED"
+	}
+}