add support for signals

Author: sduchesneau

cmd/dummy-blockchain/app/app.go

@@ -26,6 +26,7 @@ type Flags struct {
 	BlockRate         int
 	BlockSizeInBytes  int
 	ServerAddr        string
+	WithSignal        bool
 	Tracer            string
 	StopHeight        uint64
 
@@ -50,7 +51,6 @@ func Main(version string) {
 		logrus.Fatal(err)
 	}
 
-
 	root.AddCommand(
 		makeInitCommand(),
 		makeResetCommand(),
@@ -77,6 +77,7 @@ func initFlags(root *cobra.Command) error {
 	flags.Uint64Var(&cliOpts.StopHeight, "stop-height", 0, "Stop block production at this height")
 	flags.StringVar(&cliOpts.ServerAddr, "server-addr", "0.0.0.0:8080", "Server address")
 	flags.StringVar(&cliOpts.Tracer, "tracer", "", "The tracer to use, either <empty>, none or firehose")
+	flags.BoolVar(&cliOpts.WithSignal, "with-signal", false, "whether we produce BlockCommitmentLevel signals on top of blocks")
 
 	return nil
 }
@@ -166,6 +167,7 @@ func makeStartComand() *cobra.Command {
 				cliOpts.StopHeight,
 				cliOpts.ServerAddr,
 				blockTracer,
+				cliOpts.WithSignal,
 			)
 
 			if err := node.Initialize(); err != nil {

core/engine.go

@@ -19,6 +19,7 @@ type Engine struct {
 	blockSizeInBytes  int
 	blockRate         time.Duration
 	blockChan         chan *types.Block
+	signalChan        chan *types.Signal
 	prevBlock         *types.Block
 	finalBlock        *types.Block
 }
@@ -35,6 +36,7 @@ func NewEngine(genesisHash string, genesisHeight uint64, genesisTime time.Time,
 		blockRate:         blockRate,
 		blockSizeInBytes:  blockSizeInBytes,
 		blockChan:         make(chan *types.Block),
+		signalChan:        make(chan *types.Signal),
 	}
 }
 
@@ -49,7 +51,7 @@ func (e *Engine) Initialize(prevBlock *types.Block, finalBlock *types.Block) err
 	return nil
 }
 
-func (e *Engine) StartBlockProduction(ctx context.Context) {
+func (e *Engine) StartBlockProduction(ctx context.Context, withSignal bool) {
 	ticker := time.NewTicker(e.blockRate)
 
 	logrus.WithField("rate", e.blockRate).Info("starting block producer")
@@ -60,18 +62,47 @@ func (e *Engine) StartBlockProduction(ctx context.Context) {
 		}
 	}
 
+	var lastBlock *types.Block
+	var lastSignal *types.Signal
+
+	signalTicker := time.NewTicker(e.blockRate)
+	if withSignal {
+		<-time.After(e.blockRate / 2) // offset by half duration
+		signalTicker.Reset(e.blockRate)
+	} else {
+		signalTicker.Stop()
+	}
+
 	for {
 		select {
 		case <-ticker.C:
 			for _, block := range e.createBlocks() {
 				e.blockChan <- block
+				lastBlock = block
 
 				if e.stopHeight > 0 && block.Header.Height >= e.stopHeight {
 					logrus.Info("reached stop height")
 					ticker.Stop()
 					return
 				}
 			}
+		case <-signalTicker.C:
+			if !withSignal {
+				continue // just ignore if a signal ticker comes in, but it actually should not be called because of the Stop(), unless there is a crazy race condtition
+			}
+			if lastBlock != nil {
+				if lastSignal != nil && lastSignal.BlockID == lastBlock.Header.Hash {
+					continue // don't send duplicate signal
+				}
+				sig := &types.Signal{
+					BlockID:         lastBlock.Header.Hash,
+					BlockNumber:     lastBlock.Header.Height,
+					CommitmentLevel: 10,
+				}
+				e.signalChan <- sig
+				lastSignal = sig
+			}
+
 		case <-ctx.Done():
 			logrus.Info("stopping block producer")
 			close(e.blockChan)
@@ -80,10 +111,14 @@ func (e *Engine) StartBlockProduction(ctx context.Context) {
 	}
 }
 
-func (e *Engine) Subscription() <-chan *types.Block {
+func (e *Engine) SubscribeBlocks() <-chan *types.Block {
 	return e.blockChan
 }
 
+func (e *Engine) SubscribeSignals() <-chan *types.Signal {
+	return e.signalChan
+}
+
 func (e *Engine) createBlocks() (out []*types.Block) {
 	if e.prevBlock == nil {
 		genesisBlock := types.GenesisBlock(e.genesisHash, e.genesisHeight, e.genesisTime)

core/node.go

@@ -12,10 +12,11 @@ import (
 )
 
 type Node struct {
-	engine Engine
-	server Server
-	store  *Store
-	tracer tracer.Tracer
+	engine     Engine
+	server     Server
+	store      *Store
+	tracer     tracer.Tracer
+	withSignal bool
 }
 
 func NewNode(
@@ -29,14 +30,16 @@ func NewNode(
 	stopHeight uint64,
 	serverAddr string,
 	tracer tracer.Tracer,
+	withSignal bool,
 ) *Node {
 	store := NewStore(storeDir, genesisHash, genesisHeight, genesisTime)
 
 	return &Node{
-		engine: NewEngine(genesisHash, genesisHeight, genesisTime, genesisBlockBurst, stopHeight, blockRate, blockSizeInBytes),
-		store:  store,
-		server: NewServer(store, serverAddr),
-		tracer: tracer,
+		engine:     NewEngine(genesisHash, genesisHeight, genesisTime, genesisBlockBurst, stopHeight, blockRate, blockSizeInBytes),
+		store:      store,
+		server:     NewServer(store, serverAddr),
+		tracer:     tracer,
+		withSignal: withSignal,
 	}
 }
 
@@ -97,11 +100,11 @@ func (node *Node) Initialize() error {
 
 func (node *Node) Start(ctx context.Context) error {
 	go node.server.Start() // TODO: handle error here
-	go node.engine.StartBlockProduction(ctx)
+	go node.engine.StartBlockProduction(ctx, node.withSignal)
 
 	for {
 		select {
-		case block, ok := <-node.engine.Subscription():
+		case block, ok := <-node.engine.SubscribeBlocks():
 			if !ok {
 				return nil
 			}
@@ -127,6 +130,15 @@ func (node *Node) Start(ctx context.Context) error {
 				tracer.OnBlockEnd(block, node.engine.finalBlock.Header)
 			}
 
+		case sig, ok := <-node.engine.SubscribeSignals():
+			if !ok {
+				return nil
+			}
+
+			if tracer := node.tracer; tracer != nil {
+				tracer.OnCommitmentSignal(sig)
+			}
+
 		case <-ctx.Done():
 			return nil
 		}

tracer/firehose_tracer.go

@@ -75,12 +75,20 @@ func (t *FirehoseTracer) OnBlockStart(header *types.BlockHeader) {
 		},
 	}
 
-	if header.PrevHash != nil {
+	if header.PrevHash != nil && header.PrevNum != nil {
 		t.activeBlock.Header.PreviousNum = header.PrevNum
 		t.activeBlock.Header.PreviousHash = header.PrevHash
 	}
 }
 
+func (t *FirehoseTracer) OnCommitmentSignal(sig *types.Signal) {
+	fmt.Printf("FIRE SIGNAL 1 %d %s %d\n",
+		sig.BlockNumber,
+		sig.BlockID,
+		sig.CommitmentLevel,
+	)
+}
+
 // OnTrxStart implements Tracer.
 func (t *FirehoseTracer) OnTrxStart(trx *types.Transaction) {
 	if t.activeTrx != nil {

tracer/tracer.go

@@ -9,6 +9,8 @@ type Tracer interface {
 
 	OnBlockStart(header *types.BlockHeader)
 
+	OnCommitmentSignal(sig *types.Signal)
+
 	OnTrxStart(trx *types.Transaction)
 
 	OnTrxEvent(trxHash string, event *types.Event)

types/types.go

@@ -52,6 +52,12 @@ type Block struct {
 	Transactions []Transaction `json:"transactions"`
 }
 
+type Signal struct {
+	BlockID         string
+	BlockNumber     uint64
+	CommitmentLevel int32
+}
+
 // ApproximatedSize computes an approximation of how big the block would be
 // once converted into a Protobuf Block model
 func (b *Block) ApproximatedSize() int {