Added features, logging, and comments:

- Simulation returns written messages.
- sbcp/sequencer sealed block var renaming.
- Proof timeout config logic
- Proof timeout resets seq number to 0

compose/compose.go

@@ -61,3 +61,16 @@ const (
 	DecisionStateAccepted
 	DecisionStateRejected
 )
+
+func (d *DecisionState) String() string {
+	switch *d {
+	case DecisionStatePending:
+		return "Pending"
+	case DecisionStateAccepted:
+		return "Accepted"
+	case DecisionStateRejected:
+		return "Rejected"
+	default:
+		return "Unknown"
+	}
+}

compose/go.mod

@@ -6,6 +6,10 @@ require github.com/stretchr/testify v1.9.0
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
+	github.com/mattn/go-isatty v0.0.19 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/rs/zerolog v1.34.0 // indirect
+	golang.org/x/sys v0.12.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

compose/go.sum

@@ -1,9 +1,24 @@
+github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
+github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
+github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
+github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
+github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA=
+github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0=
+github.com/rs/zerolog v1.34.0 h1:k43nTLIwcTVQAncfCw4KZ2VY6ukYoZaBPNOE8txlOeY=
+github.com/rs/zerolog v1.34.0/go.mod h1:bJsvje4Z08ROH4Nhs5iH600c3IkWhwp44iRc54W6wYQ=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.12.0 h1:CM0HF96J0hcLAwsHPJZjfdNzs0gftsLfgKt57wWHJ0o=
+golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

compose/sbcp/publisher.go

@@ -18,7 +18,7 @@ var (
 
 type Publisher interface {
 	// StartPeriod is called whenever a new period starts (i.e. CurrEthereumEpoch % 10 == 0).
-	StartPeriod(newPeriodID compose.PeriodID) error
+	StartPeriod() error
 	// StartInstance is called by the upper layer to try starting a new instance from the queued requests.
 	StartInstance(req []compose.Transaction) (compose.Instance, error)
 	// DecideInstance is called once an instance gets decided.
@@ -35,7 +35,6 @@ type Publisher interface {
 
 type Messenger interface {
 	BroadcastStartPeriod(periodID compose.PeriodID, targetSuperblockNumber compose.SuperblockNumber)
-	BroadcastStartInstance(instance compose.Instance)
 	BroadcastRollback(
 		superblockNumber compose.SuperblockNumber,
 		superblockHash compose.SuperBlockHash,
@@ -54,6 +53,10 @@ type PublisherState struct {
 	SequenceNumber compose.SequenceNumber   // Per-period sequence counter (monotone)
 	ActiveChains   map[compose.ChainID]bool // Chains with active instances
 
+	// Proof window duration (in number of superblocks/periods) through which a pending superblock can be proven.
+	// StartPeriods are rejected if the next superblock is bigger than LastFinalizedSuperblockNumber + ProofWindow.
+	// 0 value means no window constrain.
+	ProofWindow uint64
 }
 
 type publisher struct {
@@ -62,18 +65,22 @@ type publisher struct {
 	PublisherState
 }
 
+// NewPublisher creates a new Publisher instance given a config, a period ID and the last settled state.
+// TargetSuperblockNumber is set to lastFinalizedSuperblockNumber initially.
+// StartPeriod needs to be called to start the first period, automatically incrementing PeriodID and TargetSuperblockNumber.
+// Thus, if the current period is N, call NewPublisher with periodID = N-1.
 func NewPublisher(messenger Messenger,
 	periodID compose.PeriodID,
-	targetSuperblockNumber compose.SuperblockNumber,
 	lastFinalizedSuperblockNumber compose.SuperblockNumber,
 	lastFinalizedSuperblockHash compose.SuperBlockHash,
+	proofWindow uint64,
 ) Publisher {
 	return &publisher{
 		mu:        sync.Mutex{},
 		messenger: messenger,
 		PublisherState: PublisherState{
 			PeriodID:               periodID,
-			TargetSuperblockNumber: targetSuperblockNumber,
+			TargetSuperblockNumber: lastFinalizedSuperblockNumber,
 
 			// Last finalized state in L1
 			LastFinalizedSuperblockNumber: lastFinalizedSuperblockNumber,
@@ -82,30 +89,39 @@ func NewPublisher(messenger Messenger,
 			// Instances scheduling
 			SequenceNumber: 0,
 			ActiveChains:   make(map[compose.ChainID]bool),
+
+			ProofWindow: proofWindow,
 		},
 	}
 }
 
 // StartPeriod is called whenever a new period starts (i.e. CurrEthereumEpoch % 10 == 0).
-func (p *publisher) StartPeriod(newPeriodID compose.PeriodID) error {
+// SequenceNumber is reset to 0, though ActiveChains is kept because instances may exist through the boundary until finished.
+func (p *publisher) StartPeriod() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	if p.TargetSuperblockNumber != p.LastFinalizedSuperblockNumber+1 {
+	// Proof window constrain
+	// If the oldest pending superblock is older than ProofWindow, reject starting the new period
+	// as the upper layer should have called ProofTimeout already.
+	nextSuperblock := p.TargetSuperblockNumber + 1
+	if nextSuperblock > p.LastFinalizedSuperblockNumber+compose.SuperblockNumber(1+p.ProofWindow) {
 		return fmt.Errorf("target superblock is %d, expected %d: %w",
 			p.TargetSuperblockNumber, p.LastFinalizedSuperblockNumber+1, ErrCannotStartPeriod)
 	}
 
-	p.PeriodID = newPeriodID
-	p.TargetSuperblockNumber++
+	p.PeriodID++
+	p.TargetSuperblockNumber = nextSuperblock
 
 	p.messenger.BroadcastStartPeriod(p.PeriodID, p.TargetSuperblockNumber)
 
 	p.SequenceNumber = 0
 	return nil
 }
 
-// StartInstance loops through the queued XT requests and tries one instance if possible.
+// StartInstance is called by the upper layer to try starting a new instance.
+// If the instance can not be started, it returns an error.
+// Else, it returns the created instance.
 func (p *publisher) StartInstance(request []compose.Transaction) (compose.Instance, error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -124,7 +140,6 @@ func (p *publisher) StartInstance(request []compose.Transaction) (compose.Instan
 	// Create instance
 	p.SequenceNumber++
 	instance := compose.Instance{
-		// TODO: generate unique ID
 		ID: generateInstanceID(
 			p.PeriodID,
 			p.SequenceNumber,
@@ -138,8 +153,6 @@ func (p *publisher) StartInstance(request []compose.Transaction) (compose.Instan
 	for _, chainID := range chains {
 		p.ActiveChains[chainID] = true
 	}
-	// Broadcast
-	p.messenger.BroadcastStartInstance(instance)
 	return instance, nil
 }
 
@@ -150,18 +163,21 @@ func (p *publisher) DecideInstance(instance compose.Instance) error {
 
 	chains := instance.Chains()
 
+	// Check instance chains are really active
 	for _, chainID := range chains {
 		if _, ok := p.ActiveChains[chainID]; !ok {
 			return ErrChainNotActive
 		}
 	}
 
+	// Remove chains from active
 	for _, chainID := range chains {
 		delete(p.ActiveChains, chainID)
 	}
 	return nil
 }
 
+// AdvanceSettledState is called when L1 emits a new settled state event
 func (p *publisher) AdvanceSettledState(
 	superblockNumber compose.SuperblockNumber,
 	superblockHash compose.SuperBlockHash,
@@ -177,11 +193,14 @@ func (p *publisher) AdvanceSettledState(
 	return nil
 }
 
+// ProofTimeout is called whenever a pending superblock is not proven within the allowed proof window.
+// It triggers a rollback to the last finalized superblock, resetting the active chains and sequence number.
 func (p *publisher) ProofTimeout() {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
 	p.ActiveChains = make(map[compose.ChainID]bool)
+	p.SequenceNumber = 0
 	p.TargetSuperblockNumber = p.LastFinalizedSuperblockNumber + 1
 	p.messenger.BroadcastRollback(p.LastFinalizedSuperblockNumber, p.LastFinalizedSuperblockHash)
 }

compose/sbcp/sbcp.go

@@ -8,8 +8,6 @@ import (
 	"github.com/compose-network/specs/compose"
 )
 
-type PriorityQueue [][]compose.Transaction
-
 // generateInstanceID returns SHA256(periodID || seq || req[0].Bytes() || ... || req[n].Bytes())
 func generateInstanceID(
 	periodID compose.PeriodID,

compose/sbcp/sequencer.go

@@ -2,8 +2,9 @@ package sbcp
 
 import (
 	"errors"
-	"github.com/compose-network/specs/compose"
 	"sync"
+
+	"github.com/compose-network/specs/compose"
 )
 
 var (
@@ -28,19 +29,19 @@ type Sequencer interface {
 		currentPeriodID compose.PeriodID,
 	) (BlockHeader, error)
 
-	// AdvancedSettledState is called when the L1 settlement event has occurred.
+	// AdvanceSettledState is called when the L1 settlement event has occurred.
 	AdvanceSettledState(SettledState)
 
-	// Block builder:
-	// Called at start of a new block
+	// Block builder policy
+	// BeginBlock is called at start of a new block
 	BeginBlock(blockNumber BlockNumber) error
-	// Return whether a local tx is admissible right now.
+	// CanIncludeLocalTx return whether a local tx is admissible right now.
 	CanIncludeLocalTx() (include bool, err error)
-	// SCP start-up hook. Locks local txs from being added (internal logic).
+	// OnStartInstance is an SCP start-up hook. Locks local txs from being added (internal logic).
 	OnStartInstance(id compose.InstanceID) error
-	// SCP decision hook. Unlocks local txs (internal logic).
+	// OnDecidedInstance is an SCP decision hook. Unlocks local txs (internal logic).
 	OnDecidedInstance(id compose.InstanceID) error
-	// End-of-block hook
+	// EndBlock: hook for when block ends
 	EndBlock(b BlockHeader) error
 }
 
@@ -61,8 +62,8 @@ type SequencerState struct {
 	// Head represents the highest sealed block number.
 	Head BlockNumber
 
-	SealedBlocks map[compose.PeriodID]SealedBlockHeader
-	SettledState SettledState
+	SealedBlockHead map[compose.PeriodID]SealedBlockHeader
+	SettledState    SettledState
 }
 
 type sequencer struct {
@@ -86,7 +87,7 @@ func NewSequencer(
 			PendingBlock:           nil,
 			ActiveInstanceID:       nil,
 			Head:                   settledState.BlockHeader.Number,
-			SealedBlocks:           make(map[compose.PeriodID]SealedBlockHeader),
+			SealedBlockHead:        make(map[compose.PeriodID]SealedBlockHeader),
 			SettledState:           settledState,
 		},
 	}
@@ -107,7 +108,7 @@ func (s *sequencer) StartPeriod(
 	// Else, it can be triggered right away.
 	if s.PendingBlock == nil {
 		var header *BlockHeader
-		block, ok := s.SealedBlocks[s.PeriodID-1]
+		block, ok := s.SealedBlockHead[s.PeriodID-1]
 		if ok {
 			header = &block.BlockHeader
 		}
@@ -116,7 +117,7 @@ func (s *sequencer) StartPeriod(
 	return nil
 }
 
-// BeginBlock is a hook called at the start of a new block.
+// BeginBlock is a hook called at the start of a new L2 block.
 func (s *sequencer) BeginBlock(blockNumber BlockNumber) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -183,7 +184,7 @@ func (s *sequencer) EndBlock(b BlockHeader) error {
 		return ErrBlockSealMismatch
 	}
 
-	s.SealedBlocks[s.PendingBlock.PeriodID] = SealedBlockHeader{
+	s.SealedBlockHead[s.PendingBlock.PeriodID] = SealedBlockHeader{
 		BlockHeader:      b,
 		PeriodID:         s.PendingBlock.PeriodID,
 		SuperblockNumber: s.PendingBlock.SuperblockNumber,
@@ -193,7 +194,7 @@ func (s *sequencer) EndBlock(b BlockHeader) error {
 	// therefore it's time to request proofs for it.
 	if s.PendingBlock.PeriodID < s.PeriodID {
 		var header *BlockHeader
-		block, ok := s.SealedBlocks[s.PeriodID-1]
+		block, ok := s.SealedBlockHead[s.PeriodID-1]
 		if ok {
 			header = &block.BlockHeader
 		}
@@ -215,7 +216,8 @@ func (s *sequencer) AdvanceSettledState(settledBlock SettledState) {
 	s.SettledState = settledBlock
 }
 
-// Rollback erases blocks beyond the given superblock number and hash, and returns the safe head block header.
+// Rollback message is sent by the publisher to all sequencers.
+// The sequencer must erase blocks beyond the given superblock number and hash, and return the safe head block header.
 func (s *sequencer) Rollback(
 	superblockNumber compose.SuperblockNumber,
 	superblockHash compose.SuperBlockHash,
@@ -228,11 +230,12 @@ func (s *sequencer) Rollback(
 	}
 
 	// Discard blocks with superblock number greater than the finalized one.
-	for bn, taggedBlock := range s.SealedBlocks {
-		if taggedBlock.SuperblockNumber > s.SettledState.SuperblockNumber {
-			delete(s.SealedBlocks, bn)
+	for blockPeriodID, sealedBlock := range s.SealedBlockHead {
+		if sealedBlock.SuperblockNumber > s.SettledState.SuperblockNumber {
+			delete(s.SealedBlockHead, blockPeriodID)
 		}
 	}
+
 	// Discard current block and active instance
 	s.PendingBlock = nil
 	s.ActiveInstanceID = nil