fix(dot/network): split stored streams and handshakeData into inbound and outbound

[noot]

Changes

• split handshakeData into inbound and outbound, since after further research into the networking, if we receive an inbound stream, the only message we send over it is the handshake. we only send messages (that aren't a handshake) over outbound streams that we initiate
• previously the code assumed that we would need to read from the outbound stream after the handshake (this was incorrect, sorry for the fake news @arijitAD )
• also previously we were only storing 1 stream but we actually needed to differentiate between inbound and outbound
• remove noGossip so we now actually gossip messages

Tests

go test ./dot/network -short

also works w/ polkadot, kusama, if you try it (and turn on trace logs) you should be some logs like:

TRCE[04-30|17:28:06] sending outbound handshake               pkg=network protocol=/dot/block-announces/1 peer=12D3KooWKjHEbPLj2ak4zCqEi2rFzreT7MQp1iNRyp3aVMjrfmh5 message="BlockAnnounceHandshake Roles=1 BestBlockNumber=3604 BestBlockHash=0xe712d655470dda112152c58b7a886b2b1a21e9565bdcbfbf4043a27cece079da GenesisHash=0x91b171bb158e2d3848fa23a9f1c25182fb8e20313b2c1eb49219da7a70ce90c3" caller=notifications.go:212
TRCE[04-30|17:28:06] sending message                          pkg=network protocol=/dot/block-announces/1 peer=12D3KooWRT7654ppFZX1mabGPNTdgDBzHY5U5ub5Xrm3eVEsntKv message="BlockAnnounceMessage ParentHash=0xb3f4bf1abcf7663ca3d7ffd3e867d05d03f2676d43a5ccf9bea61a2df309bd9b Number=4864519 StateRoot=0x02acc94df6d5a7059b14427203a0ee312f85aa9758c44a8f24e1f195290d2f41 ExtrinsicsRoot=0x1599afcabb50add4b169038f8c3a8c2ef816420a5913f373d4d9681c10bfbdf8 Digest=[PreRuntimeDigest ConsensusEngineID=BABE Data=0x03cb00000011691710000000006e1d3cc9b784f03708c7c0a10687ee45cbe27f31f1008db0d206e64466855602714bfa3ad639207f8a0623da8c37bc94830acbae0b60112cedcf71c22a2fe007591790468c29210162a5208334c638d3f06a5618ae050221260d0f17a82f1d0c SealDigest ConsensusEngineID=BABE Data=0x9c095e22771708520f72e8a4c430c1ea7291dc3ffe70b6d271871625077d8d36ba638c794d40791f786ead54e5b84a708926a387d9833de52033c89da32e478b]" caller=notifications.go:262

Issues

• related to dot/network: update notifications protocol to track specific stream w/ established handshake #1524

[noot]

this will be fixed with #1532

[codecov]

Codecov Report

Merging #1553 (fd768e9) into development (88b88f2) will decrease coverage by 0.02%.
The diff coverage is 73.56%.

@@               Coverage Diff               @@
##           development    #1553      +/-   ##
===============================================
- Coverage        58.56%   58.54%   -0.03%     
===============================================
  Files              160      160              
  Lines            16009    16001       -8     
===============================================
- Hits              9376     9367       -9     
+ Misses            4947     4946       -1     
- Partials          1686     1688       +2     

Flag	Coverage Δ
unit-tests	58.54% <73.56%> (-0.03%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
dot/network/notifications.go	62.22% <66.07%> (+0.28%)	⬆️
dot/network/sync.go	59.91% <75.00%> (-0.90%)	⬇️
dot/network/service.go	62.58% <86.36%> (+1.09%)	⬆️
dot/network/block_announce.go	73.46% <100.00%> (-4.97%)	⬇️
dot/network/test_helpers.go	81.94% <100.00%> (ø)
lib/blocktree/leaves.go	95.12% <0.00%> (+4.87%)	⬆️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 88b88f2...fd768e9. Read the comment docs.

[arijitAD]

Compute the sizeof using this.

var maxHandshakeSize = unsafe.Sizeof(BlockAnnounceHandshake{})

Also, you can store it in a constant, since size wouldn't be changing.
Add a test for it as well. The size will be 72 due to the padding of the first byte.

func TestSizeOf(t *testing.T) {
	require.Equal(t, maxHandshakeSize, uint32(72))
}

Reference: https://dlintw.github.io/gobyexample/public/memory-and-sizeof.html

[noot]

updated

[arijitAD]

Can we directly store the hsData once the handshake is sent?

[noot]

what do you mean exactly? like storing it with the stream after the data is sent?

[arijitAD]

Yes. We are storing the hsData before sending the hs with validate = false and receive = false and then adding the stream later after sending the data. I think we should update validate, receive and stream after the data is sent to the peer.

[noot]

updated to store only after sending the handshake

[arijitAD]

How long should we ignore the peer? I think a peer can still disconnect and re-initiate the handshake since we clear handshake data on close.

[noot]

yeah they can, i think we should probably add some functionality for ignoring peers that send us bad handshakes. in general we need more peer management, I can open an issue for that

[arijitAD]

There can be concurrent attempts to handshake with a peer which will lead to a race condition since we are using a network call in between. One way to fix this is by taking a lock on the handshake data.

This will ensure only one concurrent handshake call per peer.

type hsData struct {
	received  bool
	validated bool
	handshake Handshake
	stream    libp2pnetwork.Stream
	sync.Mutex
}

func (hs *hsData) String() string {
	return fmt.Sprintf(" received %t validated %t \n", hs.received, hs.validated)
}

func TestConcurrentMap(t *testing.T) {

	var m sync.Map

	task := 10

	var wg sync.WaitGroup
	wg.Add(task)

	for i := 0; i < task; i++ {
		go func() {
			defer wg.Done()
			val, ok := m.LoadOrStore("key", &hsData{
				received:  false,
				validated: false,
			})

			// There is already handshake ongoing.
			if ok {
				fmt.Println("Someone already initiated handshake")
				return
			}

			hs, _ := val.(*hsData)

			// Acquire lock so only one goroutine can access it at a time
			hs.Lock()
			defer hs.Unlock()

			// Write data to test race.
			hs.received = true
			fmt.Println(hs.String())

			hs.validated = true
			fmt.Println(hs.String())
		}()
	}
	wg.Wait()
}

[noot]

good catch, updated this function to use a lock for the hsData

[arijitAD]

@noot
In readStream to read the handshake we are sending preallocated buffer of size maxHandshakeSize which is calculated the size based on struct size. This seems incorrect since we are sending scale encoded data which may have different size and this may cause segment fault in readStream.
I can think of two ways to fix it.

1. In readStream we can send a nil buffer and once after reading the length if it is less than maxBlockResponseSize we can allocate the buffer of size(length)
2. Implement sizeof that uses scale encoding logic to figure out the size of a constant struct.

[noot]

@arijitAD the SCALE encoding of a struct is the same as the size of all the struct fields encoded and concatenated, and since BlockAnnounceHandshake has only fixed size fields, the size of a SCALE encoded BlockAnnounceHandshake is the same as the size of a non-encoded BlockAnnounceHandshake. does that fix the issue?

[arijitAD]

@arijitAD the SCALE encoding of a struct is the same as the size of all the struct fields encoded and concatenated, and since BlockAnnounceHandshake has only fixed size fields, the size of a SCALE encoded BlockAnnounceHandshake is the same as the size of a non-encoded BlockAnnounceHandshake. does that fix the issue?

Yes. Using sizeof is fine then.

[arijitAD]

This PR looks great. Just one minor change. Embed the mutex in the struct.

type handshakeData struct {
	received  bool
	validated bool
	handshake Handshake
	stream    libp2pnetwork.Stream
	*sync.Mutex
}

func newHandshakeData(received, validated bool, stream libp2pnetwork.Stream) handshakeData {
	return handshakeData{
		received:  received,
		validated: validated,
		stream:    stream,
		Mutex:     new(sync.Mutex),
	}
}

[noot]

@arijitAD nice one, I wasn't sure how to initialize the mutex without naming it :D will update!

[github-actions]

🎉 This PR is included in version 0.6.0 🎉

The release is available on GitHub release

Your semantic-release bot 📦🚀