fix: networking: call Stream.CloseWrite right after writing

[hanabi1224]

Related Issues

It fixes potential deadlock in Hello and ChainExchange p2p protocols. Below example is a minimal repro, although currently in the actual lotus code the cbor decoder will either succeed or fail fast before deadlock happens, IMHO it's a nice improvement to have.

CloseWrite sends a FIN header over yamux which notifies the reader side that stream writing has finished.

If the stream.CloseWrite() line in the below example is commented out, it will deadlock.

to run the program

1. paste it in main.go
2. run go mod tidy
3. run go mod vendor
4. run go run .

package main

import (
	"context"
	"fmt"
	"io"
	"time"

	"github.com/libp2p/go-libp2p"
	"github.com/libp2p/go-libp2p/core/network"
)

const (
	ChainExchangeProtocolID = "/fil/chain/xchg/0.0.1"
)

func main() {
	ctx := context.Background()

	h1, err := libp2p.New()
	if err != nil {
		panic(err)
	}
	defer h1.Close()
	h1.SetStreamHandler(ChainExchangeProtocolID, handleStream)

	h2, err := libp2p.New()
	if err != nil {
		panic(err)
	}
	defer h2.Close()

	ttl, _ := time.ParseDuration("24h")
	h2.Peerstore().AddAddrs(h1.ID(), h1.Addrs(), ttl)
	stream, _ := h2.NewStream(ctx, h1.ID(), ChainExchangeProtocolID)
	defer stream.Close()
	stream.Write([]byte("ping"))

	// CloseWrite sends FIN header over yamux, without it
	// the example program will deadlock
	stream.CloseWrite()

	bytes, _ := io.ReadAll(stream)
	fmt.Printf("[h2] Done reading: %s\n", string(bytes))
}

func handleStream(stream network.Stream) {
	defer stream.Close()
	fmt.Println("[h1] Start reading")
	bytes, _ := io.ReadAll(stream)
	fmt.Printf("[h1] Done reading: %s\n", string(bytes))
	stream.Write([]byte("pong"))
}

Output with CloseWrite:

[h1] Start reading
[h1] Done reading: ping
[h2] Done reading: pong

Output without CloseWrite:

[h1] Start reading

Proposed Changes

Additional Info

Checklist

Before you mark the PR ready for review, please make sure that:

• Commits have a clear commit message.
• PR title is in the form of of <PR type>: <area>: <change being made>
  • example: fix: mempool: Introduce a cache for valid signatures
  • PR type: fix, feat, build, chore, ci, docs, perf, refactor, revert, style, test
  • area, e.g. api, chain, state, market, mempool, multisig, networking, paych, proving, sealing, wallet, deps
• New features have usage guidelines and / or documentation updates in
  • Lotus Documentation
  • Discussion Tutorials
• Tests exist for new functionality or change in behavior
• CI is green

[geoff-vball]

@hanabi1224 Thanks for the contribution! @magik6k has a FIXME here, so I'm going to let him review this. He's on vacation, but we should get this in after xmas :)

[magik6k]

Thanks for the PR!

This seems to be more about the client side being nice / more correct. Reading cborrpc isn't like ReadAll, it will stop after the cbor object is fully read.

Change looks good, other than missing error check.

[magik6k]

lint fails due to missing error check

[hanabi1224]

Fixed