[RFC] core: Introduce prepare timer

Any client application could behave improperly whether or not intended,
so MinBFT cluster has to handle faulty client behaviors.
But currently replica set processes a request only when the primary
replica receives it, so any client can easily trigger view-change by
sending request messages only to all backup replicas.

This patch tries to solve the situation by introducing prepare timer,
which is started when receiving a request message and broadcasts it
to all other replicas when expired. Another existing timer, request
timer, is adjusted to cover only the latter half of consensus process.

Signed-off-by: Naoya Horiguchi <n-horiguchi@ah.jp.nec.com>

api/api.go

@@ -42,8 +42,13 @@ type Configer interface {
 	// L: must be larger than CheckpointPeriod
 	Logsize() uint32
 
-	// starts when receives a request and stops when request is accepted
+	// starts when receives a prepare message and stops when request
+	// is accepted
 	TimeoutRequest() time.Duration
+
+	// starts when receives a request and stops when request is prepared
+	TimeoutPrepare() time.Duration
+
 	// starts when sends VIEW-CHANGE and stops when receives a valid NEW-VIEW
 	TimeoutViewChange() time.Duration
 }

core/internal/clientstate/client-state.go

@@ -93,6 +93,13 @@ func NewProvider(opts ...Option) Provider {
 // has not yet expired, it will be canceled and a new timer started.
 //
 // StopRequestTimer stops timer started by StartRequestTimer, if any.
+//
+// StartPrepareTimer starts a timer to expire after the duration of
+// prepare timeout. The supplied callback function handleTimeout is
+// invoked asynchronously upon timer expiration. If the previous timer
+// has not yet expired, it will be canceled and a new timer started.
+//
+// StopPrepareTimer stops timer started by StartPrepareTimer, if any.
 type State interface {
 	CaptureRequestSeq(seq uint64) (new bool, release func())
 	PrepareRequestSeq(seq uint64) (new bool, err error)
@@ -103,6 +110,9 @@ type State interface {
 
 	StartRequestTimer(handleTimeout func())
 	StopRequestTimer()
+
+	StartPrepareTimer(handleTimeout func())
+	StopPrepareTimer()
 }
 
 // New creates a new instance of client state representation. Optional
@@ -117,6 +127,7 @@ func New(opts ...Option) State {
 	s.seqState = newSeqState()
 	s.replyState = newReplyState()
 	s.requestTimerState = newRequestTimeoutState(&s.opts)
+	s.prepareTimerState = newPrepareTimeoutState(&s.opts)
 
 	return s
 }
@@ -127,11 +138,13 @@ type Option func(*options)
 type options struct {
 	timerProvider  timer.Provider
 	requestTimeout func() time.Duration
+	prepareTimeout func() time.Duration
 }
 
 var defaultOptions = options{
 	timerProvider:  timer.Standard(),
 	requestTimeout: func() time.Duration { return time.Duration(0) },
+	prepareTimeout: func() time.Duration { return time.Duration(0) },
 }
 
 // WithTimerProvider specifies the abstract timer implementation to
@@ -151,10 +164,20 @@ func WithRequestTimeout(timeout func() time.Duration) Option {
 	}
 }
 
+// WithPrepareTimeout specifies a function that returns the duration
+// to use when starting a new prepare timeout timer. Zero or negative
+// duration disables the timeout. The timeout is disabled by default.
+func WithPrepareTimeout(timeout func() time.Duration) Option {
+	return func(opts *options) {
+		opts.prepareTimeout = timeout
+	}
+}
+
 type clientState struct {
 	*seqState
 	*replyState
 	*requestTimerState
+	*prepareTimerState
 
 	opts options
 }

core/internal/clientstate/prepare-timeout.go

@@ -0,0 +1,63 @@
+// Copyright (c) 2019 NEC Solution Innovators, Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clientstate
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/hyperledger-labs/minbft/core/internal/timer"
+)
+
+type prepareTimerState struct {
+	sync.Mutex
+
+	prepareTimer timer.Timer
+
+	opts *options
+}
+
+func newPrepareTimeoutState(opts *options) *prepareTimerState {
+	return &prepareTimerState{opts: opts}
+}
+
+func (s *prepareTimerState) StartPrepareTimer(forward func()) {
+	s.Lock()
+	defer s.Unlock()
+
+	timerProvider := s.opts.timerProvider
+	timeout := s.opts.prepareTimeout()
+
+	if s.prepareTimer != nil {
+		s.prepareTimer.Stop()
+	}
+
+	if timeout <= time.Duration(0) {
+		return
+	}
+
+	fmt.Printf("start forward timer: timeout = %d\n", timeout)
+	s.prepareTimer = timerProvider.AfterFunc(timeout, forward)
+}
+
+func (s *prepareTimerState) StopPrepareTimer() {
+	s.Lock()
+	defer s.Unlock()
+
+	if s.prepareTimer != nil {
+		s.prepareTimer.Stop()
+	}
+}