Merge pull request #180 from cloudstruct/feat/localtxsubmission-timeout

feat: make localtxsubmission timeout configurable

Author: agaffney

protocol/blockfetch/client.go

@@ -21,7 +21,7 @@ func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
 		config: cfg,
 	}
 	// Update state map with timeouts
-	stateMap := StateMap
+	stateMap := StateMap.Copy()
 	if entry, ok := stateMap[STATE_BUSY]; ok {
 		entry.Timeout = c.config.BatchStartTimeout
 		stateMap[STATE_BUSY] = entry
@@ -40,7 +40,7 @@ func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
 		Role:                protocol.ProtocolRoleClient,
 		MessageHandlerFunc:  c.messageHandler,
 		MessageFromCborFunc: NewMsgFromCbor,
-		StateMap:            StateMap,
+		StateMap:            stateMap,
 		InitialState:        STATE_IDLE,
 	}
 	c.Protocol = protocol.New(protoConfig)

protocol/chainsync/client.go

@@ -38,7 +38,7 @@ func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
 		currentTipChan:        make(chan Tip),
 	}
 	// Update state map with timeouts
-	stateMap := StateMap
+	stateMap := StateMap.Copy()
 	if entry, ok := stateMap[STATE_INTERSECT]; ok {
 		entry.Timeout = c.config.IntersectTimeout
 		stateMap[STATE_INTERSECT] = entry
@@ -59,7 +59,7 @@ func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
 		Role:                protocol.ProtocolRoleClient,
 		MessageHandlerFunc:  c.messageHandler,
 		MessageFromCborFunc: msgFromCborFunc,
-		StateMap:            StateMap,
+		StateMap:            stateMap,
 		InitialState:        STATE_IDLE,
 	}
 	c.Protocol = protocol.New(protoConfig)

protocol/handshake/client.go

@@ -19,7 +19,7 @@ func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
 		config: cfg,
 	}
 	// Update state map with timeout
-	stateMap := StateMap
+	stateMap := StateMap.Copy()
 	if entry, ok := stateMap[STATE_CONFIRM]; ok {
 		entry.Timeout = c.config.Timeout
 		stateMap[STATE_CONFIRM] = entry
@@ -34,7 +34,7 @@ func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
 		Role:                protocol.ProtocolRoleClient,
 		MessageHandlerFunc:  c.handleMessage,
 		MessageFromCborFunc: NewMsgFromCbor,
-		StateMap:            StateMap,
+		StateMap:            stateMap,
 		InitialState:        STATE_PROPOSE,
 	}
 	c.Protocol = protocol.New(protoConfig)

protocol/keepalive/client.go

@@ -21,7 +21,7 @@ func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
 		config: cfg,
 	}
 	// Update state map with timeout
-	stateMap := StateMap
+	stateMap := StateMap.Copy()
 	if entry, ok := stateMap[STATE_SERVER]; ok {
 		entry.Timeout = c.config.Timeout
 		stateMap[STATE_SERVER] = entry

protocol/localtxsubmission/client.go

@@ -15,10 +15,21 @@ type Client struct {
 }
 
 func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
+	if cfg == nil {
+		tmpCfg := NewConfig()
+		cfg = &tmpCfg
+	}
 	c := &Client{
 		config:           cfg,
 		submitResultChan: make(chan error),
 	}
+	// Update state map with timeout
+	stateMap := StateMap.Copy()
+	if entry, ok := stateMap[STATE_BUSY]; ok {
+		entry.Timeout = c.config.Timeout
+		stateMap[STATE_BUSY] = entry
+	}
+	// Configure underlying Protocol
 	protoConfig := protocol.ProtocolConfig{
 		Name:                PROTOCOL_NAME,
 		ProtocolId:          PROTOCOL_ID,
@@ -28,7 +39,7 @@ func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
 		Role:                protocol.ProtocolRoleClient,
 		MessageHandlerFunc:  c.messageHandler,
 		MessageFromCborFunc: NewMsgFromCbor,
-		StateMap:            StateMap,
+		StateMap:            stateMap,
 		InitialState:        STATE_IDLE,
 	}
 	c.Protocol = protocol.New(protoConfig)

protocol/localtxsubmission/localtxsubmission.go

@@ -1,6 +1,8 @@
 package localtxsubmission
 
 import (
+	"time"
+
 	"github.com/cloudstruct/go-ouroboros-network/protocol"
 )
 
@@ -50,6 +52,7 @@ type LocalTxSubmission struct {
 
 type Config struct {
 	SubmitTxFunc SubmitTxFunc
+	Timeout      time.Duration
 }
 
 // Callback function types
@@ -66,7 +69,9 @@ func New(protoOptions protocol.ProtocolOptions, cfg *Config) *LocalTxSubmission
 type LocalTxSubmissionOptionFunc func(*Config)
 
 func NewConfig(options ...LocalTxSubmissionOptionFunc) Config {
-	c := Config{}
+	c := Config{
+		Timeout: 30 * time.Second,
+	}
 	// Apply provided options functions
 	for _, option := range options {
 		option(&c)
@@ -79,3 +84,9 @@ func WithSubmitTxFunc(submitTxFunc SubmitTxFunc) LocalTxSubmissionOptionFunc {
 		c.SubmitTxFunc = submitTxFunc
 	}
 }
+
+func WithTimeout(timeout time.Duration) LocalTxSubmissionOptionFunc {
+	return func(c *Config) {
+		c.Timeout = timeout
+	}
+}

protocol/state.go

@@ -41,3 +41,13 @@ type StateMapEntry struct {
 }
 
 type StateMap map[State]StateMapEntry
+
+// Copy returns a copy of the state map. This is mostly for convenience,
+// since we need to copy the state map in various places
+func (s StateMap) Copy() StateMap {
+	ret := StateMap{}
+	for k, v := range s {
+		ret[k] = v
+	}
+	return ret
+}