network: new stream! protocol and pull syncer implementation

[acud]

This PR is the result of joint efforts in the core track to realise the exact, up to date requirements of the stream! protocol and implement them so that their functionality and behavior is easier to reason about.

The design borrows some code from the old stream package. Namely the way we manage bins we want to stream in the case of syncing and the existing intervals store. However everything else including tests have been built from the ground up.

The PR does not yet remove the existing stream code, does not move the intervals package, but rather builds on top of those. This is in order to decrease the strain of diffing those changes. These would be deferred to subsequent PRs after this one is merged.

The new stream implementation suggested in this PR brings more guarantees, better performance and safety to how we handle streams, namely:

• Spamming through streams is far less likely due to tight checking of wanted streams from a certain peer
• Spamming of unsolicited chunks is completely forbidden - a peer cannot send us the same chunk twice, a peer cannot send us a chunk that we did not request too (these are two different cases)
• Multiple chunks can now be delivered on the same ChunkDelivery message
• No duplicate chunks are sent between two peers

In terms of reduced complexity:

• Far less state is maintained on each side of the connection. Servers maintain offers, clients maintain wants

PLEASE NOTE: that tests constitute about a half of this PR. Some of them overlap in functionality and will be removed, namely some of the tests in cursors_test.go.

Actionable items that still need to be addressed, however I would personally like to defer to a later PR:

• Move stream maintanance for the case of syncing out of the sync provider - the sync provider needs to only sink data to the stream protocol, not manage subscriptions. This, however, needs a bit more tweaking in terms of abstraction for stream - as it would require to expose certain subset of functionalities out of the stream package. In laymen terminology - we would need to call certain functions on stream when a depth change happens for example, but then also to have hooks on stream to know that a StreamInfoRes message has arrived and what to do accordingly....
• how to guaranty that historical and live streams are kept alive - we should have metrics to see at least on the smoke test environment that these streams keep on getting requested and served correctly
• verify that existing historical and live historical stream fetchers terminate when depth changes and node moves out of depth - this is not so trivial to test but the logic that does this is in the codebase
• maxPeers - old stream package had a notion of maximum peers we'd like to stream to/with. we have concluded partially that kademlia should somehow shield us from having to implement this at the moment, since number of connections are/should be enforced by kademlia at the moment. so this needs more info before we jump the gun to implement such a requirement
• proper tracing with jaeger and opentracing - need more info from @nonsense and @skylenet on how to do this properly
• Update diagrams

HOW TO REVIEW:
Since there is a fair amount of business logic here, a brief overview on how the protocol works is needed in order to facilitate an effective review process.
Flow is as follows:

• Stream protocol and Sync Provider are registered on swarm.go. Stream handles the protocol message exchange, syncProvider handles sinking the data and managing streams according to kademlia
• Peers connect and establish protocol over devp2p. This uses the run function in the stream.go file
• Peer initialises providers
• Sync provider initialises streams according to kademlia depth (InitPeer function on sync_provider.go)
• Message exchange starts from the previous step - stream protocol requests info about wanted streams with StreamInfoReq. Server replies with info about streams (boundedness, cursors (previously known as sessionIndex) in StreamInfoRes message
• Client handles StreamInfoRes, and according to wantedness of streams, initiates GetRange queries to fetch the stream from the server.
• Server replies with OfferedHashes message for the requested range
• Client, as necessary, requests hashes from the set of OfferedHashes with a WantedHashes message
• Server delivers, as necessary, chunks wanted with ChunkDelivery message. This message can contain multiple chunks as needed. Also, theoretically ,multiple ChunkDelivery messages could be sent in reply for the same WantedHashes in order to optimise amounts of messages sent and according to db performance to sink the data
• Once all chunks for a requested range are delivered, the interval in the interval store is sealed, and the next range is requested

related to ethersphere/user-stories#2
fixes #1451
somewhat fixes #1393
closes #1457
closes #1247
closes #1246
closes #1234
closes #1204
closes #1203
closes #1105

[nonsense]

Please remove Roundtrip, I really don't want to have to submit any more PRs like #1422 and #1339

I think developers on the team are smart enough to be able to add this field once they need it.

[nonsense]

In general do not add code that is not used for no good reason. A good reason is if you will have an immediate PR that will make use of it, for example the PR with bzz-retrieval, even though the code was not immediately used is a good example.

[nonsense]

I think this is a good use-case of something we could test if we redirect logs with a simulation. Then we could check for presence of the given log message containing addr and id. For now we don't have this functionality, so fine to leave this here, but ultimately we will be able to test this kind of impl. behaviour without custom hooks.

Just an idea.

[nonsense]

I fail to see why we need Boundedness and Bound. Why do we add such indirection without using it. If this is so important how come we don't have an implementation of it.

From the spec:
when a client requests a bounded range, server should respond to the client range requests with either offered hashes (if roundtrip is required) or chunks (if not) or an end-of-batch message if there are no more to offer. If none of these responses arrive within a timeout interval, client must drop the upstream peer.

We don't have an example of Boundedness or of Roundtrip, so I fail to see why this is useful, apart from introducing indirection in already not simple functionality. We've been working on pull-sync new version for months, the previous version has been buggy in production for probably more than a year (and still throwing subscription errors), but we still continue to add code prematurely so that we make our lives harder...

[nonsense]

@zelig @janos this comment is still applicable. @janos no action required from your side, but if you understand this, maybe you can explain it to me :)

[nonsense]

You are already unhooking the old stream protocol, so please remove this code, not just commented out, and also remove all the code in network/stream that is no longer used. No point in keeping it around, specially when it is not hooked up anymore here in swarm.go, and we are directly switching to the new implementation.

[nonsense]

The mental taxation in understanding what this is doing is a bit too high for my taste. I'm happy we have started to push away complexity from production Swarm code, but we should probably rethink the complexity in tests or at least be mindful about that.

It is not immediately obvious what this test is testing. Tests should generally be simple and I shouldn't have to read the code and think for 15min to figure that out. What is this testing?

1. We are using a simulation framework.
2. We are mocking a devp2p protocol and injecting some custom hook
3. We are building peer fixtures using the p2p library.

We should probably be building more tools to write tests in a more simple manner.

No action required (or at least I don't know what I would expect), but this is def. not the direction we should be going in general IMO.

[nonsense]

@acud @janos overall great work. I don't see any blockers from my side. I've added a few comments, and talked to @acud off-band about some things that need to be fleshed out, but nothing major.

1. I still see some missing Godoc and left out commented code - it'd be nice to fix this prior to merging.
2. I think we should remove asap the stream folder with the previous version of the stream protocol - it is no longer used or hooked up, so no need to have dead code in the codebase.
3. I really think we should not add "code to be used in the future" such as Bounded streams and Roundtrip fields that serve no purpose. I've explained multiple times my opinion on that, I leave it up to you to decide.
4. Grafana dashboards are not up-to-date with the new bzz-stream protocol - please fix them - namely the P2P dashboard. I also suggest adding a bit of timers around pull-syncing batches, so that we have a baseline of how fast the pull-syncing is, and how much time it takes to request/deliver a batch. I'm sure you can figure out other interesting operational metrics that we should monitor, so that we have a baseline and improve on it. We spoke off-band about that with @acud
5. Jaeger HTTP GET traces appear to be missing the po log lines for some reason. Not a big deal, but nice to have. I tried to fix this quickly, but failed, not sure whats wrong with the OpenTracing span.
6. I am not very happy with our custom use of mocked protocols and service functions... this is one of the reasons we are building the new simulation framework that will always start a production Swarm node. I think we can write better tests in general, and by better I mean - easier to reason about, easier to understand, and easier to maintain. Right now the tests are definitely not easy to understand or read, and I don't think they are very maintainable. At the same time I understand that right now, today, we don't have these tools, so I can understand why you wrote those tests the way you did. For now at least improve on the Godoc on them to make things slightly easier for the reader.

These are my general comments. The most important thing is that the code seems to be correct and integration tests seem to be passing, so I see this generally as an improvement to master, and think we can move forward with it.

[zelig]

@acud @janos

We created a super spec for stream! and didnt implement it. instead we implement a temporary mix of sync and stream which was identified as a problem with the current implementation.

This PR saw the tension between two different approaches one that condones any abstraction in the spirit of 'just write what the thing should do now nothing more' type of simplicity on the one hand and
the other side which cherishes clearly identified and sees simplicity, approachability and testability in the correct separation of concerns in the form of the right abstractions.

I kindly ask you to consider the following simple sketchy architecture. and think how much it would simplify testing, help understanding and incrementally PR-ing this thing.

Thanks

package stream

// Range represents a closed interval
type Range struct {
	From, To uint64 // closed interval
}

// individual streams - no need for actual interface
type Stream interface {
	String() string
	Bound() bool
}

// Provider manages the source of a type of stream
// this is completely stateless
type Provider interface {
	Type() string                                                 // Name of provider, eg. SYNC
	CanHandle(Name) (lastIndex uint64, bound bool, err error) // provider can serve the stream given as argument, return cursor, boundedness and an  error if not provided
	OfferedHashes(Name, Range) []Address                          // returns the offered hashes for the given range
	GetDataByHash(Name, ...Address) []Chunk                       // retrieves the chunks based on Address
}

// Streamer implements Registry interface to be used in provider packages
//
type Registry interface {
	Register(Provider) error
}

// LocalStoreProvider wraps localstore and implements GetDataByHash call with localstore get
type LocalStoreProvider struct {
	localStore
}

func (l *LocalProvider) GetDataByHash(ctx context.Context, hash []byte) (chunk.Chunk, error) {
	return l.localStore.Get(ctx, chunk.ModeGetSync, hash)
}

// Client Logic
// Request Object - no need for interface, just to show methods
type Request interface {
	Cancel()
	Done() (close <-chan struct{}) // like in context
	Err()                          // error to extract after Done() closed
}

// Peer - this is not an interface just demonstrating the relevant methods
type Peer interface {
	GetRange(ctx context.Context, n Name, r Range, need func([]byte) bool, deliver func(Chunk) error) (Request, error)
	Providers() []Info
}

// NetStoreClient is a stream client that
// - proxies need function as localstore Has
// - puts chunk in netstore
// These clients are stateless
type StorerClient struct {
	Streamer
	intervals IntervalStore
	store     NetStore
	need      func([]byte) bool // implements it with netStore.Has and opens a fetcher
	deliver   func(Chunk) error // implements it with netStore.Put chunk.ModePutSync
}

// Request will try to cover the indicated range using interval store
// and peer requests for each gap in the intervals
func (s *StorerClient) Request(p Peer, n Name, r Range) (Request, error) {
	// starts a go routine that keeps getting the intervals based on the interval store
	// continuously requesting subranges according to the interval store
	// and calls
	p.GetRange(ctx, n, r, s.need, s.deliver)
	//  as batches are sealed, intervals are updated
	// the request is alive until the whole range is covered
}

// and now the level doing full sync of a stream
type Mode = uin64

const (
	History Mode = iota
	Live
	Both
)

// StorerClient implements Syncer, here stream.Syncer means syncing a stream
// only interface  used by the pull syncer client
type Syncer interface {
	Request(Context, Peer, Name, Mode) (Request, error)
	// this falls back to until cursor (history) and from cursor (live) range requests from the StorerClient
	//
}

////////////////////////////
// package pullsync

// pullSync.Server implements stream.Provider
// uses localstore SubscribePull for the  bins
// server is node-wide
type Server struct {
	// ...
	*stream.LocalProvider
}

// the node-wide pullsync.Client
type Client struct {
	stream.Syncer // embed stream.Syncer
	// when pullsync
	// here  you simply put the update sync logic listening to kademlia depth changes
	// and call `Request` 
        // remember the request, when no longer relevant just call request.Cancel()
}

[zelig]

ok for moral's sake lets just merge this PR after some critical changes. The rest should be addressed in subsequent PRs.

• there is a lot of changes in packages totally unrelated to this PR, (docker, pyramid chunker, etc) quite a few unnecessary
• filestore, hasherstore, localstore and intervals store changes are important and unexplained
• the package should be in network/stream/v2 and newstream should be dropped
• error handling and peer drops could be massively simplified
• select on routine termination conditions are scattered around in many unnecessary places
• the technical debt this PR introduces is way too high. Lets avoid this in future

[zelig]

comment exported functions

[zelig]

remove commented code

[zelig]

surely not in production right?

[zelig]

// space

[zelig]

I would prefer an explicit register method, but can wait till next iteration

[zelig]

upped -> upper

and better say 'we can set the open wants upper bound to the index supplied in the msg.

Maybe some checks would be needed about this number

[zelig]

why is this needed?

[zelig]

better have cache in a separate object

[zelig]

please just use the snapshots innetwork/stream/testing

[nonsense]

The idea is to remove the whole network/stream package. Why create references between the new and old packages, when we should have only one in the codebase?

[zelig]

@nonsense, no the idea is to reimplement streamer

• we want to use the same snapshots as before
• we want pr to be smaller
• bitvector and intervalstore are also dependencies from old stream
• the new streamer should be in stream/v2

[nonsense]

OK, I was unaware that we will be maintaining both streamers. cc @janos

[nonsense]

Actually @janos is already removing it at 120c034, which I thought is a good idea, but now I am a bit confused why we need to keep it around? @zelig why do you want to keep both v1 and v2? Is there another library that is using the code in v1?

[zelig]

^

[nonsense]

I am not sure why we need jq and websocat within the image. What is the use-case here? I use those tools from my local machine, why do we need them in the swarm binary?

[nonsense]

I think we should revert this. I didn't understand why we wanted to have it smaller in the first place to be honest.

[nonsense]

return nil, true, let's be explicit that o is no found.

[janos]

@zelig @nonsense Could you check this benchmark also? I think that the measuring method is not correct. The simulation is constructed outside of the benchmark loop and in the benchmark loop chunks are uploaded on the same node of the same simulation, making the state on every benchmark iteration different. Every iteration measures different state and with larger number of chunks in the localstore, it becomes slower. This makes this benchmark result dependent on the number of iterations performed, where only the precisions of results should be influenced by the number of iterations. I think that this benchmark should be rewritten in a way that every iteration measures time on the independent state of other iterations, or to remove the benchmark as that may make this benchmark time consuming.

[nonsense]

@janos it does look like the benchmark is indeed not correct for the reasons you mentioned.

[zelig]

agree

[janos]

I have changed the benchmark to address this problem, and it is even usable.

[zelig]

thanks @janos for the quick cleanup and reorg, I dont like that you put in the deletion of old stream code in this PR.

i approve it but lets make sure:

• we heed the issues raised in later PRs
• we understand the changes to localstore, intervalstore, simulation, etc packages

[janos]

@zelig Thank you for the review and approval.

I will create issues on the comments that are there after this PR gets merged.

I do not mind putting back the old streamer code. Would you like to restore the files?

[nonsense]

We can't sync 32 chunks ? :)

[janos]

It times out on travis. :) And it actually does not even flake locally. The only thing that I managed to see is the goroutine leak in hive that is fixed. If it is flaky even with only 4 chunks on travis, we should disable this test and dedicate more time to find the root cause.