integrate header exchange service to rollkit nodes (cosmos#684)

Integrating header exchange service to rollkit nodes using go-header
package. Fixes rollkit/rollkit#674 and
fixes rollkit/rollkit#672

block/manager.go

@@ -56,8 +56,9 @@ type Manager struct {
 	// daHeight is the height of the latest processed DA block
 	daHeight uint64
 
-	HeaderOutCh chan *types.SignedHeader
-	HeaderInCh  chan *types.SignedHeader
+	HeaderOutCh     chan *types.SignedHeader
+	HeaderInCh      chan *types.SignedHeader
+	SyncedHeadersCh chan *types.SignedHeader
 
 	lastCommit atomic.Value
 
@@ -138,13 +139,14 @@ func NewManager(
 		retriever:   dalc.(da.BlockRetriever), // TODO(tzdybal): do it in more gentle way (after MVP)
 		daHeight:    s.DAHeight,
 		// channels are buffered to avoid blocking on input/output operations, buffer sizes are arbitrary
-		HeaderOutCh:    make(chan *types.SignedHeader, 100),
-		HeaderInCh:     make(chan *types.SignedHeader, 100),
-		blockInCh:      make(chan newBlockEvent, 100),
-		FraudProofInCh: make(chan *abci.FraudProof, 100),
-		retrieveMtx:    new(sync.Mutex),
-		syncCache:      make(map[uint64]*types.Block),
-		logger:         logger,
+		HeaderOutCh:     make(chan *types.SignedHeader, 100),
+		HeaderInCh:      make(chan *types.SignedHeader, 100),
+		SyncedHeadersCh: make(chan *types.SignedHeader, 100),
+		blockInCh:       make(chan newBlockEvent, 100),
+		FraudProofInCh:  make(chan *abci.FraudProof, 100),
+		retrieveMtx:     new(sync.Mutex),
+		syncCache:       make(map[uint64]*types.Block),
+		logger:          logger,
 	}
 	agg.retrieveCond = sync.NewCond(agg.retrieveMtx)
 
@@ -216,6 +218,7 @@ func (m *Manager) SyncLoop(ctx context.Context, cancel context.CancelFunc) {
 			m.retrieveCond.Signal()
 		case header := <-m.HeaderInCh:
 			m.logger.Debug("block header received", "height", header.Header.Height(), "hash", header.Header.Hash())
+			m.SyncedHeadersCh <- header
 			newHeight := header.Header.BaseHeader.Height
 			currentHeight := m.store.Height()
 			// in case of client reconnecting after being offline

config/config.go

@@ -20,6 +20,7 @@ const (
 	flagNamespaceID   = "rollkit.namespace_id"
 	flagFraudProofs   = "rollkit.experimental_insecure_fraud_proofs"
 	flagLight         = "rollkit.light"
+	flagTrustedHash   = "rollkit.trusted_hash"
 )
 
 // NodeConfig stores Rollkit node configuration.
@@ -35,6 +36,12 @@ type NodeConfig struct {
 	DALayer            string `mapstructure:"da_layer"`
 	DAConfig           string `mapstructure:"da_config"`
 	Light              bool   `mapstructure:"light"`
+	HeaderConfig       `mapstructure:",squash"`
+}
+
+// HeaderConfig allows node to pass the initial trusted header hash to start the header exchange service
+type HeaderConfig struct {
+	TrustedHash string `mapstructure:"trusted_hash"`
 }
 
 // BlockManagerConfig consists of all parameters required by BlockManagerConfig
@@ -67,6 +74,7 @@ func (nc *NodeConfig) GetViperConfig(v *viper.Viper) error {
 		return err
 	}
 	copy(nc.NamespaceID[:], bytes)
+	nc.TrustedHash = v.GetString(flagTrustedHash)
 	return nil
 }
 
@@ -84,4 +92,5 @@ func AddFlags(cmd *cobra.Command) {
 	cmd.Flags().BytesHex(flagNamespaceID, def.NamespaceID[:], "namespace identifies (8 bytes in hex)")
 	cmd.Flags().Bool(flagFraudProofs, def.FraudProofs, "enable fraud proofs (experimental & insecure)")
 	cmd.Flags().Bool(flagLight, def.Light, "run light client")
+	cmd.Flags().String(flagTrustedHash, def.TrustedHash, "initial trusted hash to start the header exchange service")
 }

config/defaults.go

@@ -27,4 +27,7 @@ var DefaultNodeConfig = NodeConfig{
 	DALayer:  "mock",
 	DAConfig: "",
 	Light:    false,
+	HeaderConfig: HeaderConfig{
+		TrustedHash: "",
+	},
 }

go.mod

@@ -35,6 +35,7 @@ require (
 	github.com/benbjohnson/clock v1.3.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/btcsuite/btcd v0.22.1 // indirect
+	github.com/celestiaorg/go-libp2p-messenger v0.1.0 // indirect
 	github.com/cespare/xxhash v1.1.0 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/containerd/cgroups v1.0.4 // indirect
@@ -55,6 +56,8 @@ require (
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/go-kit/log v0.2.1 // indirect
 	github.com/go-logfmt/logfmt v0.5.1 // indirect
+	github.com/go-logr/logr v1.2.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-resty/resty/v2 v2.7.0 // indirect
 	github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
@@ -85,7 +88,6 @@ require (
 	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
 	github.com/jbenet/goprocess v0.1.4 // indirect
 	github.com/jmhodges/levigo v1.0.0 // indirect
-	github.com/jtolds/gls v4.20.0+incompatible // indirect
 	github.com/klauspost/compress v1.15.12 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.1 // indirect
 	github.com/koron/go-ssdp v0.0.3 // indirect
@@ -154,6 +156,8 @@ require (
 	github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 // indirect
 	go.etcd.io/bbolt v1.3.6 // indirect
 	go.opencensus.io v0.24.0 // indirect
+	go.opentelemetry.io/otel v1.11.1 // indirect
+	go.opentelemetry.io/otel/trace v1.11.1 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/dig v1.15.0 // indirect
 	go.uber.org/fx v1.18.2 // indirect