Log incoming subscription messages.

stream/stream.go

@@ -135,6 +135,8 @@ func subscribe(ctx context.Context, log log.Logger, ps PubSub, handle func(conte
 	s := ps.Subscribe(ctx)
 	defer s.Close()
 
+	log = log.WithField("topic", ps.String())
+
 	for {
 		msg, err := s.Next(ctx)
 		if err != nil {
@@ -146,10 +148,12 @@ func subscribe(ctx context.Context, log log.Logger, ps PubSub, handle func(conte
 				return
 			}
 
-			log.WithError(err).
-				WithField("topic", ps.String()).
+			log.With(msg).
+				WithError(err).
 				Warn("failed to handle subscription event")
 		}
+
+		log.With(msg).Debug("handled subscription event")
 	}
 }
 

stream/transport/transport.go

@@ -29,12 +29,37 @@ type Subscription interface {
 
 type ForkVersionFormat uint64
 
+func (f ForkVersionFormat) String() string {
+	switch f {
+	case Unknown:
+		return "none"
+	case AltairJson:
+		return "altair/json"
+	case BellatrixJson:
+		return "bellatrix/json"
+	case CapellaJson:
+		return "capella/json"
+	case CapellaSSZ:
+		return "capella/ssz"
+	}
+
+	return fmt.Sprintf("invalid: %d", f)
+}
+
 type Message struct {
 	Source       uuid.UUID
 	ForkEncoding ForkVersionFormat
 	Payload      []byte
 }
 
+func (m Message) Loggable() map[string]any {
+	return map[string]any{
+		"source":        m.Source,
+		"fork_encoding": m.ForkEncoding,
+		"n_bytes":       len(m.Payload),
+	}
+}
+
 func Encode(m Message) ([]byte, error) {
 	rawItem, err := json.Marshal(m)
 	if err != nil {