*: store parsigdb metrics

app/app.go

@@ -567,7 +567,32 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 		return err
 	}
 
-	parSigDB := parsigdb.NewMemDB(lock.Threshold, deadlinerFunc("parsigdb"))
+	var (
+		slotDuration uint64
+		genesisTime  time.Time
+	)
+
+	if conf.TestnetConfig.IsNonZero() {
+		slotDuration = 12
+		genesisTime = time.Unix(conf.TestnetConfig.GenesisTimestamp, 0)
+	} else {
+		network, err := eth2util.ForkVersionToNetwork(lock.ForkVersion)
+		if err != nil {
+			network = "mainnet"
+		}
+
+		slotDuration, err = eth2util.NetworkToSlotDuration(network)
+		if err != nil {
+			return err
+		}
+
+		genesisTime, err = eth2util.NetworkToGenesisTime(network)
+		if err != nil {
+			return err
+		}
+	}
+
+	parSigDB := parsigdb.NewMemDB(lock.Threshold, deadlinerFunc("parsigdb"), parsigdb.NewMemDBMetadata(slotDuration, genesisTime))
 
 	var parSigEx core.ParSigEx
 	if conf.TestConfig.ParSigExFunc != nil {

core/dutydb/memory.go

@@ -34,7 +34,6 @@ func NewMemDB(deadliner core.Deadliner) *MemDB {
 }
 
 // MemDB is an in-memory dutyDB implementation.
-// It is a placeholder for the badgerDB implementation.
 type MemDB struct {
 	mu sync.Mutex
 

core/parsigdb/memory.go

@@ -6,26 +6,41 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"strconv"
 	"sync"
+	"time"
 
 	"github.com/obolnetwork/charon/app/errors"
 	"github.com/obolnetwork/charon/app/log"
 	"github.com/obolnetwork/charon/app/z"
 	"github.com/obolnetwork/charon/core"
 )
 
+// NewMemDBMetadata returns a new in-memory partial signature database instance.
+func NewMemDBMetadata(slotDuration uint64, genesisTime time.Time) MemDBMetadata {
+	return MemDBMetadata{
+		slotDuration: slotDuration,
+		genesisTime:  genesisTime,
+	}
+}
+
+type MemDBMetadata struct {
+	slotDuration uint64
+	genesisTime  time.Time
+}
+
 // NewMemDB returns a new in-memory partial signature database instance.
-func NewMemDB(threshold int, deadliner core.Deadliner) *MemDB {
+func NewMemDB(threshold int, deadliner core.Deadliner, metadata MemDBMetadata) *MemDB {
 	return &MemDB{
 		entries:    make(map[key][]core.ParSignedData),
 		keysByDuty: make(map[core.Duty][]key),
 		threshold:  threshold,
 		deadliner:  deadliner,
+		metadata:   metadata,
 	}
 }
 
-// MemDB is a placeholder in-memory partial signature database.
-// It will be replaced with a BadgerDB implementation.
+// MemDB is an in-memory partial signature database.
 type MemDB struct {
 	mu           sync.Mutex
 	internalSubs []func(context.Context, core.Duty, core.ParSignedDataSet) error
@@ -35,6 +50,8 @@ type MemDB struct {
 	keysByDuty map[core.Duty][]key
 	threshold  int
 	deadliner  core.Deadliner
+
+	metadata MemDBMetadata
 }
 
 // SubscribeInternal registers a callback when an internal
@@ -146,6 +163,21 @@ func (db *MemDB) store(k key, value core.ParSignedData) ([]core.ParSignedData, b
 	db.mu.Lock()
 	defer db.mu.Unlock()
 
+	now := time.Now().UnixMilli()
+
+	slotStart := (uint64(db.metadata.genesisTime.Unix()) + k.Duty.Slot*db.metadata.slotDuration) * 1000 // in ms
+	timeSinceSlotStart := float64(now-int64(slotStart)) / 1000                                          // in seconds
+
+	switch k.Duty.Type {
+	case core.DutyAttester:
+		timeSinceSlotStart -= 4.0
+	case core.DutyAggregator, core.DutySyncContribution:
+		timeSinceSlotStart -= 8.0
+	default:
+	}
+
+	parsigStored.WithLabelValues(k.Duty.Type.String(), strconv.FormatInt(int64(value.ShareIdx), 10)).Observe(timeSinceSlotStart)
+
 	for _, s := range db.entries[k] {
 		if s.ShareIdx == value.ShareIdx {
 			equal, err := parSignedDataEqual(s, value)

core/parsigdb/memory_internal_test.go

@@ -5,6 +5,7 @@ package parsigdb
 import (
 	"context"
 	"testing"
+	"time"
 
 	eth2v1 "github.com/attestantio/go-eth2-client/api/v1"
 	"github.com/attestantio/go-eth2-client/spec/altair"
@@ -13,6 +14,7 @@ import (
 
 	"github.com/obolnetwork/charon/cluster"
 	"github.com/obolnetwork/charon/core"
+	"github.com/obolnetwork/charon/eth2util"
 	"github.com/obolnetwork/charon/testutil"
 )
 
@@ -115,7 +117,7 @@ func TestMemDBThreshold(t *testing.T) {
 	)
 
 	deadliner := newTestDeadliner()
-	db := NewMemDB(th, deadliner)
+	db := NewMemDB(th, deadliner, NewMemDBMetadata(eth2util.Mainnet.SlotDuration, time.Unix(eth2util.Mainnet.GenesisTimestamp, 0)))
 
 	ctx := t.Context()
 

core/parsigdb/metrics.go

@@ -14,3 +14,11 @@ var exitCounter = promauto.NewCounterVec(prometheus.CounterOpts{
 	Name:      "exit_total",
 	Help:      "Total number of partially signed voluntary exits per public key",
 }, []string{"pubkey"}) // Ok to use pubkey (high cardinality) here since these are very rare
+
+var parsigStored = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Namespace: "core",
+	Subsystem: "parsigdb",
+	Name:      "store",
+	Help:      "Latency of partial signatures received since earliest expected time, per duty, per peer index",
+	Buckets:   []float64{.001, 0.01, 0.05, .1, .25, .5, .75, 1, 1.25, 1.5, 1.75, 2.0, 2.25, 2.5, 2.75, 3, 5},
+}, []string{"duty", "peer_idx"})

dkg/exchanger.go

@@ -82,7 +82,7 @@ func newExchanger(p2pNode host.Host, peerIdx int, peers []peer.ID, sigTypes []si
 
 	ex := &exchanger{
 		// threshold is len(peers) to wait until we get all the partial sigs from all the peers per DV
-		sigdb:    parsigdb.NewMemDB(len(peers), noopDeadliner{}),
+		sigdb:    parsigdb.NewMemDB(len(peers), noopDeadliner{}, parsigdb.NewMemDBMetadata(0, time.Now())), // metadata timestamps are used for metrics, irrelevant for DKG
 		sigex:    parsigex.NewParSigEx(p2pNode, p2p.Send, peerIdx, peers, noopVerifier, dutyGaterFunc, p2p.WithSendTimeout(timeout), p2p.WithReceiveTimeout(timeout)),
 		sigTypes: st,
 		sigData: dataByPubkey{

dkg/pedersen/reshare.go

@@ -124,6 +124,7 @@ func RunReshareDKG(ctx context.Context, config *Config, board *Board, shares []s
 		for _, removedPeerID := range config.Reshare.RemovedPeers {
 			if idx, ok := config.PeerMap[removedPeerID]; ok && idx.PeerIdx == int(node.Index) {
 				isRemoving = true
+
 				if idx.PeerIdx == thisNodeIndex {
 					thisIsRemovedNode = true
 				}
@@ -136,6 +137,7 @@ func RunReshareDKG(ctx context.Context, config *Config, board *Board, shares []s
 		for _, addedPeerID := range config.Reshare.AddedPeers {
 			if idx, ok := config.PeerMap[addedPeerID]; ok && idx.PeerIdx == int(node.Index) {
 				isNewlyAdded = true
+
 				if idx.PeerIdx == thisNodeIndex {
 					thisIsAddedNode = true
 				}

docs/metrics.md

@@ -56,6 +56,7 @@ when storing metrics from multiple nodes or clusters in one Prometheus instance.
 | `core_consensus_timeout_total` | Counter | Total count of consensus timeouts by protocol, duty, and timer | `protocol, duty, timer` |
 | `core_fetcher_proposal_blinded` | Gauge | Whether the fetched proposal was blinded (1) or local (2) |  |
 | `core_parsigdb_exit_total` | Counter | Total number of partially signed voluntary exits per public key | `pubkey` |
+| `core_parsigdb_store` | Histogram | Latency of partial signatures received since earliest expected time, per duty, per peer index | `duty, peer_idx` |
 | `core_scheduler_current_epoch` | Gauge | The current epoch |  |
 | `core_scheduler_current_slot` | Gauge | The current slot |  |
 | `core_scheduler_duty_total` | Counter | The total count of duties scheduled by type | `duty` |

eth2util/network.go

@@ -27,6 +27,8 @@ type Network struct {
 	GenesisTimestamp int64
 	// CapellaHardFork represents capella fork version, used for computing domains for signatures
 	CapellaHardFork string
+	// SlotDuration represents slot duration in seconds
+	SlotDuration uint64
 }
 
 // IsNonZero checks if each field in this struct is not equal to its zero value.
@@ -42,34 +44,39 @@ var (
 		GenesisForkVersionHex: "0x00000000",
 		GenesisTimestamp:      1606824023,
 		CapellaHardFork:       "0x03000000",
+		SlotDuration:          12,
 	}
 	Goerli = Network{
 		ChainID:               5,
 		Name:                  "goerli",
 		GenesisForkVersionHex: "0x00001020",
 		GenesisTimestamp:      1616508000,
 		CapellaHardFork:       "0x03001020",
+		SlotDuration:          12,
 	}
 	Gnosis = Network{
 		ChainID:               100,
 		Name:                  "gnosis",
 		GenesisForkVersionHex: "0x00000064",
 		GenesisTimestamp:      1638993340,
 		CapellaHardFork:       "0x03000064",
+		SlotDuration:          5,
 	}
 	Chiado = Network{
 		ChainID:               10200,
 		Name:                  "chiado",
 		GenesisForkVersionHex: "0x0000006f",
 		GenesisTimestamp:      1665396300,
 		CapellaHardFork:       "0x0300006f",
+		SlotDuration:          5,
 	}
 	Sepolia = Network{
 		ChainID:               11155111,
 		Name:                  "sepolia",
 		GenesisForkVersionHex: "0x90000069",
 		GenesisTimestamp:      1655733600,
 		CapellaHardFork:       "0x90000072",
+		SlotDuration:          12,
 	}
 	// Holesky metadata taken from https://github.com/eth-clients/holesky#metadata.
 	Holesky = Network{
@@ -78,6 +85,7 @@ var (
 		GenesisForkVersionHex: "0x01017000",
 		GenesisTimestamp:      1696000704,
 		CapellaHardFork:       "0x04017000",
+		SlotDuration:          12,
 	}
 	// Hoodi metadata taken from https://github.com/eth-clients/hoodi/#metadata.
 	Hoodi = Network{
@@ -86,6 +94,7 @@ var (
 		GenesisForkVersionHex: "0x10000910",
 		GenesisTimestamp:      1742213400,
 		CapellaHardFork:       "0x40000910",
+		SlotDuration:          12,
 	}
 )
 
@@ -192,6 +201,15 @@ func NetworkToGenesisTime(name string) (time.Time, error) {
 	return time.Unix(network.GenesisTimestamp, 0), nil
 }
 
+func NetworkToSlotDuration(name string) (uint64, error) {
+	network, err := networkFromName(name)
+	if err != nil {
+		return 0, err
+	}
+
+	return network.SlotDuration, nil
+}
+
 func ForkVersionToGenesisTime(forkVersion []byte) (time.Time, error) {
 	network, err := networkFromForkVersion(fmt.Sprintf("%#x", forkVersion))
 	if err != nil {