Merge remote-tracking branch 'origin' into jasmina/8219-blocksync-spec

.github/workflows/linkchecker.yml

@@ -1,12 +0,0 @@
-name: Check Markdown links
-on:
-  schedule:
-    - cron: '* */24 * * *'
-jobs:
-  markdown-link-check:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v3
-      - uses: creachadair/github-action-markdown-link-check@master
-        with:
-          folder-path: "docs"

abci/client/socket_client.go

@@ -17,12 +17,6 @@ import (
 	"github.com/tendermint/tendermint/libs/service"
 )
 
-const (
-	// reqQueueSize is the max number of queued async requests.
-	// (memory: 256MB max assuming 1MB transactions)
-	reqQueueSize = 256
-)
-
 // This is goroutine-safe, but users should beware that the application in
 // general is not meant to be interfaced with concurrent callers.
 type socketClient struct {
@@ -48,7 +42,7 @@ var _ Client = (*socketClient)(nil)
 func NewSocketClient(logger log.Logger, addr string, mustConnect bool) Client {
 	cli := &socketClient{
 		logger:      logger,
-		reqQueue:    make(chan *requestAndResponse, reqQueueSize),
+		reqQueue:    make(chan *requestAndResponse),
 		mustConnect: mustConnect,
 		addr:        addr,
 		reqSent:     list.New(),
@@ -118,6 +112,11 @@ func (cli *socketClient) sendRequestsRoutine(ctx context.Context, conn io.Writer
 		case <-ctx.Done():
 			return
 		case reqres := <-cli.reqQueue:
+			// N.B. We must enqueue before sending out the request, otherwise the
+			// server may reply before we do it, and the receiver will fail for an
+			// unsolicited reply.
+			cli.trackRequest(reqres)
+
 			if err := types.WriteMessage(reqres.Request, bw); err != nil {
 				cli.stopForError(fmt.Errorf("write to buffer: %w", err))
 				return
@@ -158,14 +157,15 @@ func (cli *socketClient) recvResponseRoutine(ctx context.Context, conn io.Reader
 	}
 }
 
-func (cli *socketClient) willSendReq(reqres *requestAndResponse) {
-	cli.mtx.Lock()
-	defer cli.mtx.Unlock()
-
+func (cli *socketClient) trackRequest(reqres *requestAndResponse) {
+	// N.B. We must NOT hold the client state lock while checking this, or we
+	// may deadlock with shutdown.
 	if !cli.IsRunning() {
 		return
 	}
 
+	cli.mtx.Lock()
+	defer cli.mtx.Unlock()
 	cli.reqSent.PushBack(reqres)
 }
 
@@ -199,7 +199,6 @@ func (cli *socketClient) doRequest(ctx context.Context, req *types.Request) (*ty
 	}
 
 	reqres := makeReqRes(req)
-	cli.willSendReq(reqres)
 
 	select {
 	case cli.reqQueue <- reqres:

internal/consensus/reactor.go

@@ -216,6 +216,8 @@ func (r *Reactor) OnStart(ctx context.Context) error {
 		if err := r.state.Start(ctx); err != nil {
 			return err
 		}
+	} else if err := r.state.updateStateFromStore(); err != nil {
+		return err
 	}
 
 	go r.updateRoundStateRoutine(ctx)

internal/consensus/state.go

@@ -122,9 +122,8 @@ type State struct {
 	// store blocks and commits
 	blockStore sm.BlockStore
 
-	stateStore            sm.Store
-	initialStatePopulated bool
-	skipBootstrapping     bool
+	stateStore        sm.Store
+	skipBootstrapping bool
 
 	// create and execute blocks
 	blockExec *sm.BlockExecutor
@@ -248,9 +247,6 @@ func NewState(
 }
 
 func (cs *State) updateStateFromStore() error {
-	if cs.initialStatePopulated {
-		return nil
-	}
 	state, err := cs.stateStore.Load()
 	if err != nil {
 		return fmt.Errorf("loading state: %w", err)
@@ -259,6 +255,15 @@ func (cs *State) updateStateFromStore() error {
 		return nil
 	}
 
+	eq, err := state.Equals(cs.state)
+	if err != nil {
+		return fmt.Errorf("comparing state: %w", err)
+	}
+	// if the new state is equivalent to the old state, we should not trigger a state update.
+	if eq {
+		return nil
+	}
+
 	// We have no votes, so reconstruct LastCommit from SeenCommit.
 	if state.LastBlockHeight > 0 {
 		cs.reconstructLastCommit(state)
@@ -266,7 +271,6 @@ func (cs *State) updateStateFromStore() error {
 
 	cs.updateToState(state)
 
-	cs.initialStatePopulated = true
 	return nil
 }
 

internal/proxy/client_test.go

@@ -58,7 +58,7 @@ func (app *appConnTest) Info(ctx context.Context, req *types.RequestInfo) (*type
 var SOCKET = "socket"
 
 func TestEcho(t *testing.T) {
-	sockPath := fmt.Sprintf("unix:///tmp/echo_%v.sock", tmrand.Str(6))
+	sockPath := fmt.Sprintf("unix://%s/echo_%v.sock", t.TempDir(), tmrand.Str(6))
 	logger := log.NewNopLogger()
 	client, err := abciclient.NewClient(logger, sockPath, SOCKET, true)
 	if err != nil {
@@ -98,7 +98,7 @@ func TestEcho(t *testing.T) {
 
 func BenchmarkEcho(b *testing.B) {
 	b.StopTimer() // Initialize
-	sockPath := fmt.Sprintf("unix:///tmp/echo_%v.sock", tmrand.Str(6))
+	sockPath := fmt.Sprintf("unix://%s/echo_%v.sock", b.TempDir(), tmrand.Str(6))
 	logger := log.NewNopLogger()
 	client, err := abciclient.NewClient(logger, sockPath, SOCKET, true)
 	if err != nil {
@@ -146,7 +146,7 @@ func TestInfo(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	sockPath := fmt.Sprintf("unix:///tmp/echo_%v.sock", tmrand.Str(6))
+	sockPath := fmt.Sprintf("unix://%s/echo_%v.sock", t.TempDir(), tmrand.Str(6))
 	logger := log.NewNopLogger()
 	client, err := abciclient.NewClient(logger, sockPath, SOCKET, true)
 	if err != nil {