bzz protocol first stab

bzz/protocol.go

@@ -5,3 +5,333 @@ BZZ implements the bzz wire protocol of swarm
 routing decoded storage and retrieval requests to DPA
 and registering peers with the DHT
 */
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/ethereum/go-ethereum/p2p"
+)
+
+const (
+	Version            = 0
+	ProtocolLength     = uint64(8)
+	ProtocolMaxMsgSize = 10 * 1024 * 1024
+	NetworkId          = 0
+	strategy           = 0
+)
+
+// bzz protocol message codes
+const (
+	StatusMsg          = iota // 0x01
+	StoreRequestMsg           // 0x02
+	RetrieveRequestMsg        // 0x03
+	GetBlockHashesMsg         // 0x04
+	PeersMsg                  // 0x05
+	DeliveryMsg               // 0x06
+)
+
+type BzzHive interface {
+	AddStoreRequest(*StoreRequestMsgData, *p2p.Peer) error
+	AddRetrieveRequest(*RetrieveRequestMsgData, *p2p.Peer) error
+	AddPeers([]*p2p.Peer) error
+	AddDelivery(*DeliveryMsgData, *p2p.Peer) error
+	RemovePeer(*p2p.Peer)
+}
+
+// bzzProtocol represents the swarm wire protocol
+// instance is running on each peer
+type bzzProtocol struct {
+	Hive BzzHive
+	DHT  *DHTStore
+	// CAD *p2p.Cademlia
+	peer *p2p.Peer
+	id   string
+	rw   p2p.MsgReadWriter
+}
+
+/*
+ message structs used for rlp decoding
+Handshake
+
+[0x01, Version: B_32, strategy: B_32, capacity: B_64, peers: B_8]
+
+Storing
+
+[+0x02, key: B_256, metadata: [], data: B_4k]: the data chunk to be stored, preceded by its key.
+
+Retrieving
+
+[0x03, key: B_256, timeout: B_64, metadata: []]: key of the data chunk to be retrieved, timeout in milliseconds. Note that zero timeout retrievals serve also as messages to retrieve peers.
+
+Peers
+
+[0x04, key: B_256, timeout: B_64, peers: [[peer], [peer], .... ]] the encoding of a peer is identical to that in the devp2p base protocol peers messages: [IP, Port, NodeID] note that a node's DPA address is not the NodeID but the hash of the NodeID. Timeout serves to indicate whether the responder is forwarding the query within the timeout or not.
+
+Delivery
+
+[0x05, key: B_256, metadata: [], data: B_4k]: the delivery response to retrieval queries.
+*/
+
+type StatusMsgData struct {
+	Version   uint64
+	ID        string
+	NodeID    []byte
+	NetworkId uint64
+	Caps      []p2p.Cap
+	Strategy  uint64
+}
+
+/*
+ Given the chunker I see absolutely no reason why not allow storage and delivery of larger data . See my discussion on flexible chunking.
+ store requests are forwarded to the peers in their cademlia proximity bin if they are distant
+ if they are within our storage radius or have any incentive to store it then attach your nodeID to the metadata
+ if the storage request is sufficiently close (within our proximity range (the last row of the routing table), then sending it to all peers will not guarantee convergence, so there needs to be an absolute expiry of the request too. Maybe the protocol should specify a forward probability exponentially declining with age.
+*/
+type StoreRequestMsgData struct {
+	Key    Key           // hash of datasize | data
+	Size   uint64        // size of data in bytes
+	Data   []byte        // is this needed?
+	Reader SectionReader // is the underlying byte slice already buffered within the app somewhere?
+	// optional
+	RequestTimeout time.Time // expiry for forwarding
+	StorageTimeout time.Time // expiry of content
+	Metadata       MetaData
+}
+
+/*
+Root key retrieve request
+Timeout in milliseconds. Note that zero timeout retrieval requests do not request forwarding, but prompt for a peers message response. therefore they also serve also as messages to retrieve peers.
+MaxSize specifies the maximum size that the peer will accept. This is useful in particular if we allow storage and delivery of multichunk payload representing the entire or partial subtree unfolding from the requested root key. So when only interested in limited part of a stream (infinite trees) or only testing chunk availability etc etc, we can indicate it by limiting the size here.
+In the special case that the key is identical to the peers own address (hash of NodeID) the message is to be handled as a self lookup. The response is a PeersMsg with the peers in the cademlia proximity bin corresponding to the address.
+It is unclear if a retrieval request with an empty target is the same as a self lookup
+*/
+type RetrieveRequestMsgData struct {
+	Key      Key
+	MaxSize  uint64    // optional maximum size of delivery accepted
+	Timeout  time.Time //optional, if missing or
+	Metadata MetaData
+}
+
+/* one response to retrieval, always encouraged after a retrieval request to respond with a list of peers in the same cademlia proximity bin.
+The encoding of a peer is identical to that in the devp2p base protocol peers messages: [IP, Port, NodeID]
+note that a node's DPA address is not the NodeID but the hash of the NodeID.
+Timeout serves to indicate whether the responder is forwarding the query within the timeout or not.
+The Key is the target (if response to a retrieval request) or peers address (hash of NodeID) if retrieval request was a self lookup.
+It is unclear if PeersMsg with an empty Key has a special meaning or just mean the same as with the peers address as Key (cademlia bin)
+*/
+type PeersMsgData struct {
+	Peers []*p2p.Peer
+	Key   Key // if a response to a retrieval request
+}
+
+/*
+Delivery and storeRequest messages could be lumped together or maybe distinguished by their request timeout ?
+*/
+// wonder if we should use Chunk here directly or keep loosely coupled
+type DeliveryMsgData struct {
+	Key      Key
+	Data     SectionReader
+	Metadata MetaData
+}
+
+/*
+ metadata is as yet a placeholder
+ it will likely contain info about hops or the entire forward chain of node IDs
+ this may allow some interesting schemes to evolve optimal routing strategies
+  metadata for storage and retrieval requests could specify format parameters relevant for the (blockhashing) chunking scheme used (for chunks corresponding to a treenode). For instance all runtime params for the chunker (hashing algorithm used, branching etc.)
+  Finally metadata can hold info relevant to some reward or compensation scheme that may be used to incentivise peers.
+*/
+type MetaData struct{}
+
+// main entrypoint, wrappers starting a server running the bzz protocol
+// use this constructor to attach the protocol ("class") to server caps
+// the Dev p2p layer then runs the protocol instance on each peer
+func BzzProtocol(hive BzzHive) p2p.Protocol {
+	return p2p.Protocol{
+		Name:    "bzz",
+		Version: Version,
+		Length:  ProtocolLength,
+		Run: func(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
+			return runBzzProtocol(hive, peer, rw)
+		},
+	}
+}
+
+// the main loop that handles incoming messages
+// note RemovePeer in the post-disconnect hook
+func runBzzProtocol(hive BzzHive, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) {
+	self := &bzzProtocol{
+		Hive: hive,
+		// DHT: dht,
+		// CAD:  cad,
+		rw:   rw,
+		peer: peer,
+		id:   fmt.Sprintf("%x", peer.Identity().Pubkey()[:8]),
+	}
+	err = self.handleStatus()
+	if err == nil {
+		for {
+			err = self.handle()
+			if err != nil {
+				self.Hive.RemovePeer(self.peer)
+				break
+			}
+		}
+	}
+	return
+}
+
+func (self *bzzProtocol) handle() error {
+	msg, err := self.rw.ReadMsg()
+	if err != nil {
+		return err
+	}
+	if msg.Size > ProtocolMaxMsgSize {
+		return self.protoError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
+	}
+	// make sure that the payload has been fully consumed
+	defer msg.Discard()
+	/*
+	   StatusMsg          = iota // 0x01
+	   StoreRequestMsg           // 0x02
+	   RetrieveRequestMsg        // 0x03
+	   PeersMsg                  // 0x04
+	   DeliveryMsg               // 0x05
+	*/
+	switch msg.Code {
+	case StatusMsg:
+		return self.protoError(ErrExtraStatusMsg, "")
+
+	case StoreRequestMsg:
+		var req StoreRequestMsgData
+		if err := msg.Decode(&req); err != nil {
+			return self.protoError(ErrDecode, "msg %v: %v", msg, err)
+		}
+		self.Hive.AddStoreRequest(&req, self.peer)
+
+	case RetrieveRequestMsg:
+		var req RetrieveRequestMsgData
+		if err := msg.Decode(&req); err != nil {
+			return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
+		}
+		self.Hive.AddRetrieveRequest(&req, self.peer)
+
+	case PeersMsg:
+		var req PeersMsgData
+		if err := msg.Decode(&req); err != nil {
+			return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
+		}
+		self.Hive.AddPeers(req.Peers)
+
+	case DeliveryMsg:
+		var req DeliveryMsgData
+		if err := msg.Decode(&req); err != nil {
+			return self.protoError(ErrDecode, "->msg %v: %v", msg, err)
+		}
+		self.Hive.AddDelivery(&req, self.peer)
+
+	default:
+		return self.protoError(ErrInvalidMsgCode, "%v", msg.Code)
+	}
+	return nil
+}
+
+/*
+type StatusMsg struct {
+  Version  uint64
+  ID       string
+  NodeID   []byte
+  Caps     []Cap
+  Strategy uint64
+}
+*/
+
+func (self *bzzProtocol) statusMsg() p2p.Msg {
+
+	return p2p.NewMsg(StatusMsg,
+		uint32(Version),
+		uint32(NetworkId),
+		"honey",
+		[]p2p.Cap{},
+		strategy,
+	)
+}
+
+func (self *bzzProtocol) handleStatus() error {
+	// send precanned status message
+	if err := self.rw.WriteMsg(self.statusMsg()); err != nil {
+		return err
+	}
+
+	// read and handle remote status
+	msg, err := self.rw.ReadMsg()
+	if err != nil {
+		return err
+	}
+
+	if msg.Code != StatusMsg {
+		return self.protoError(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, StatusMsg)
+	}
+
+	if msg.Size > ProtocolMaxMsgSize {
+		return self.protoError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize)
+	}
+
+	var status StatusMsgData
+	if err := msg.Decode(&status); err != nil {
+		return self.protoError(ErrDecode, "msg %v: %v", msg, err)
+	}
+
+	if status.NetworkId != NetworkId {
+		return self.protoError(ErrNetworkIdMismatch, "%d (!= %d)", status.NetworkId, NetworkId)
+	}
+
+	if Version != status.Version {
+		return self.protoError(ErrVersionMismatch, "%d (!= %d)", status.Version, Version)
+	}
+
+	self.peer.Infof("Peer is [bzz] capable (%d/%d)\n", status.Version, status.NetworkId)
+
+	self.Hive.AddPeers([]*p2p.Peer{self.peer})
+
+	return nil
+}
+
+func (self *bzzProtocol) Retrieve(req RetrieveRequestMsgData) error {
+	return p2p.EncodeMsg(self.rw, RetrieveRequestMsg, req)
+}
+
+func (self *bzzProtocol) Store(req StoreRequestMsgData) error {
+	return p2p.EncodeMsg(self.rw, StoreRequestMsg, req)
+}
+
+func (self *bzzProtocol) DeliveryMsg(req DeliveryMsgData) error {
+	return p2p.EncodeMsg(self.rw, DeliveryMsg, req)
+}
+
+func (self *bzzProtocol) Peers(req PeersMsgData) error {
+	return p2p.EncodeMsg(self.rw, PeersMsg, req)
+}
+
+func (self *bzzProtocol) protoError(code int, format string, params ...interface{}) (err *protocolError) {
+	err = ProtocolError(code, format, params...)
+	if err.Fatal() {
+		self.peer.Errorln("err %v", err)
+		// disconnect
+	} else {
+		self.peer.Debugf("fyi %v", err)
+	}
+	return
+}
+
+func (self *bzzProtocol) protoErrorDisconnect(code int, format string, params ...interface{}) {
+	err := ProtocolError(code, format, params...)
+	if err.Fatal() {
+		self.peer.Errorln("err %v", err)
+		// disconnect
+	} else {
+		self.peer.Debugf("fyi %v", err)
+	}
+
+}