Start with the store code. Properly handle announcement requests.

Add core data functions: Data2Hash, GetData, GetDataLocal, GetDataDHT, StoreDataLocal, StoreDataDHT
2026-07-18 19:17:51 +01:00 · 2021-04-10 16:00:43 +02:00
parent 96213d4ed9
commit ff9804bf6d
7 changed files with 151 additions and 24 deletions
--- a/Commands.go
+++ b/Commands.go
@@ -28,8 +28,6 @@ func (peer *PeerInfo) cmdAnouncement(msg *MessageAnnouncement) {
 		}

 		fmt.Printf("Incoming initial announcement from %s\n", msg.connection.Address.String())
-	} else {
-		fmt.Printf("Incoming secondary announcement from %s\n", msg.connection.Address.String())
 	}

 	// Filter function to only share peers that are "connectable" to the remote one. It checks IPv4, IPv6, and local connection.
@@ -43,6 +41,8 @@ func (peer *PeerInfo) cmdAnouncement(msg *MessageAnnouncement) {
 	allowIPv6 := msg.Features&(1<<FeatureIPv6Listen) > 0

 	var hash2Peers []Hash2Peer
+	var hashesNotFound [][]byte
+	var filesEmbed []EmbeddedFileData

 	// FIND_SELF: Requesting peers close to the sender?
 	if msg.Actions&(1<<ActionFindSelf) > 0 {
@@ -55,7 +55,11 @@ func (peer *PeerInfo) cmdAnouncement(msg *MessageAnnouncement) {
 			}
 		}

-		hash2Peers = append(hash2Peers, selfD)
+		if len(selfD.Closest) > 0 {
+			hash2Peers = append(hash2Peers, selfD)
+		} else {
+			hashesNotFound = append(hashesNotFound, peer.NodeID)
+		}
 	}

 	// FIND_PEER: Find a different peer?
@@ -69,22 +73,35 @@ func (peer *PeerInfo) cmdAnouncement(msg *MessageAnnouncement) {
 				}
 			}

-			hash2Peers = append(hash2Peers, details)
+			if len(details.Closest) > 0 {
+				hash2Peers = append(hash2Peers, details)
+			} else {
+				hashesNotFound = append(hashesNotFound, findPeer.Hash)
+			}
 		}
 	}

 	// Find a value?
 	if msg.Actions&(1<<ActionFindValue) > 0 {
-		// TODO query store
+		for _, findHash := range msg.FindDataKeys {
+			stored, data := announcementGetData(findHash.Hash)
+			if stored && len(data) > 0 {
+				filesEmbed = append(filesEmbed, EmbeddedFileData{ID: findHash, Data: data})
+			} else if stored {
+				selfRecord := selfPeerRecord(msg.connection.Network)
+				hash2Peers = append(hash2Peers, Hash2Peer{ID: findHash, Storing: []PeerRecord{selfRecord}})
+			} else {
+				hashesNotFound = append(hashesNotFound, findHash.Hash)
+			}
+		}
 	}

 	// Information about files stored by the sender?
-	if msg.Actions&(1<<ActionInfoStore) > 0 {
-		// TODO
+	if msg.Actions&(1<<ActionInfoStore) > 0 && len(msg.InfoStoreFiles) > 0 {
+		peer.announcementStore(msg.InfoStoreFiles)
 	}

-	// Empty announcement from existing peer means the peer most likely restarted. For regular connection upkeep ping should be used.
-	peer.sendResponse(added, hash2Peers, nil, nil)
+	peer.sendResponse(added, hash2Peers, filesEmbed, hashesNotFound)
 }

 func (peer *PeerInfo) peer2Record(allowLocal, allowIPv4, allowIPv6 bool) (result *PeerRecord) {
@@ -151,8 +168,6 @@ func (peer *PeerInfo) cmdResponse(msg *MessageResponse) {
 			}
 		}
 	}*/
-
-	//fmt.Printf("Incoming response from %s on %s\n", msg.connection.Address.String(), msg.connection.Address.String())
 }

 // cmdPing handles an incoming ping message
--- a/Kademlia.go
+++ b/Kademlia.go
@@ -8,6 +8,7 @@ package core

 import (
 	"bytes"
+	"time"

 	"github.com/PeernetOfficial/core/dht"
 )
@@ -24,8 +25,8 @@ func initKademlia() {
 	}

 	// SendRequestStore sends a store message to the remote node. I.e. asking it to store the given key-value
-	nodesDHT.SendRequestStore = func(node *dht.Node, key []byte, value []byte) {
-		node.Info.(*PeerInfo).sendAnnouncementStore(key, value)
+	nodesDHT.SendRequestStore = func(node *dht.Node, key []byte, dataSize uint64) {
+		node.Info.(*PeerInfo).sendAnnouncementStore(key, dataSize)
 	}

 	// SendRequestFindNode sends an information request to find a particular node. nodes are the nodes to send the request to.
@@ -65,6 +66,49 @@ func (peer *PeerInfo) sendAnnouncementFindValue(request *dht.InformationRequest)
 	peer.sendAnnouncement(false, findSelf, findPeer, findValue, nil)
 }

-func (peer *PeerInfo) sendAnnouncementStore(key []byte, value []byte) {
-	peer.sendAnnouncement(false, false, nil, nil, []InfoStore{{ID: KeyHash{Hash: key}, Size: uint64(len(value)), Type: 0}})
+func (peer *PeerInfo) sendAnnouncementStore(fileHash []byte, fileSize uint64) {
+	peer.sendAnnouncement(false, false, nil, nil, []InfoStore{{ID: KeyHash{Hash: fileHash}, Size: fileSize, Type: 0}})
+}
+
+// ---- CORE DATA FUNCTIONS ----
+
+// Data2Hash returns the hash for the data
+func Data2Hash(data []byte) (hash []byte) {
+	return hashData(data)
+}
+
+// GetData returns the requested data. It checks first the local store and then tries via DHT.
+func GetData(hash []byte) (data []byte, found bool) {
+	if data, found = GetDataLocal(hash); found {
+		return data, found
+	}
+
+	return GetDataDHT(hash)
+}
+
+// GetDataLocal returns data from the local warehouse.
+func GetDataLocal(hash []byte) (data []byte, found bool) {
+	return Warehouse.Retrieve(hash)
+}
+
+// GetDataDHT requests data via DHT
+func GetDataDHT(hash []byte) (data []byte, found bool) {
+	data, found, _ = nodesDHT.Get(hash)
+	return data, found
+}
+
+// StoreDataLocal stores data into the local warehouse.
+func StoreDataLocal(data []byte) error {
+	key := hashData(data)
+	return Warehouse.Store(key, data, time.Time{}, time.Time{})
+}
+
+// StoreDataDHT stores data locally and informs peers in the DHT about it.
+// Remote peers may choose to keep a record (in case another peers asks) or mirror the full data.
+func StoreDataDHT(data []byte) error {
+	key := hashData(data)
+	if err := Warehouse.Store(key, data, time.Time{}, time.Time{}); err != nil {
+		return err
+	}
+	return nodesDHT.Store(key, uint64(len(data)))
 }
--- a/ID.go
+++ b/ID.go
@@ -69,6 +69,11 @@ func ExportPrivateKey() (privateKey *btcec.PrivateKey, publicKey *btcec.PublicKe
 	return peerPrivateKey, peerPublicKey
 }

+// SelfNodeID returns the node ID used for DHT
+func SelfNodeID() []byte {
+	return nodeID
+}
+
 // PeerInfo stores information about a single remote peer
 type PeerInfo struct {
 	PublicKey          *btcec.PublicKey // Public key
@@ -183,3 +188,14 @@ func records2Nodes(records []PeerRecord, network *Network) (nodes []*dht.Node) {

 	return
 }
+
+// selfPeerRecord returns self as peer record
+func selfPeerRecord(network *Network) (result PeerRecord) {
+	return PeerRecord{
+		PublicKey:   peerPublicKey,
+		NodeID:      nodeID,
+		IP:          network.address.IP,
+		Port:        uint16(network.address.Port),
+		LastContact: 0,
+	}
+}
--- a/Peernet.go
+++ b/Peernet.go
@@ -14,6 +14,7 @@ func Init() {
 	initBroadcastIPv4()
 	initNetwork()
 	initSeedList()
+	initStore()
 }

 // Connect starts bootstrapping and local peer discovery.
--- a/Store.go
+++ b/Store.go
@@ -0,0 +1,50 @@
+/*
+File Name:  Store.go
+Copyright:  2021 Peernet s.r.o.
+Author:     Peter Kleissner
+*/
+
+package core
+
+import (
+	"encoding/hex"
+	"fmt"
+
+	"github.com/PeernetOfficial/core/store"
+)
+
+// Warehouse contains all key-value data served via DHT
+var Warehouse store.Store
+
+// TODO: Via descriptors, files stored by other peers
+
+func initStore() {
+	Warehouse = store.NewMemoryStore()
+}
+
+// announcementGetData returns data for an announcement
+func announcementGetData(hash []byte) (stored bool, data []byte) {
+	// TODO: Create RetrieveIfSize to prevent files larger than EmbeddedFileSizeMax from being loaded
+	data, found := Warehouse.Retrieve(hash)
+	if !found {
+		return false, nil
+	}
+
+	if len(data) <= EmbeddedFileSizeMax {
+		return true, data
+	}
+
+	return true, nil
+}
+
+// announcementStore handles an incoming announcement by another peer about storing data
+func (peer *PeerInfo) announcementStore(records []InfoStore) {
+	// TODO: Only store the other peers data if certain conditions are met:
+	// - enough storage available
+	// - not exceeding record count per peer
+	// - not exceeding total record count limit
+	// - not exceeding record count per CIDR
+	for _, record := range records {
+		fmt.Printf("Remote node %s stores hash %s (size %d type %d)\n", hex.EncodeToString(peer.NodeID), hex.EncodeToString(record.ID.Hash), record.Size, record.Type)
+	}
+}
--- a/go.mod
+++ b/go.mod
@@ -3,11 +3,11 @@ module github.com/PeernetOfficial/core
 go 1.16

 require (
-	github.com/PeernetOfficial/core/dht v0.0.0-20210409140720-c44e57a0a02b
-	github.com/PeernetOfficial/core/reuseport v0.0.0-20210409140720-c44e57a0a02b
+	github.com/PeernetOfficial/core/dht v0.0.0-20210409204108-27ec231e1475
+	github.com/PeernetOfficial/core/reuseport v0.0.0-20210409204108-27ec231e1475
 	github.com/btcsuite/btcd v0.21.0-beta.0.20210401013323-36a96f6a0025
 	golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2
-	golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4
+	golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
 	lukechampine.com/blake3 v1.1.5
 )
--- a/go.sum
+++ b/go.sum
@@ -1,7 +1,7 @@
-github.com/PeernetOfficial/core/dht v0.0.0-20210409140720-c44e57a0a02b h1:4V7y4Kul7R57EAWxr85fVHG3/Gf6U6RBL3ZQFqjjSpU=
-github.com/PeernetOfficial/core/dht v0.0.0-20210409140720-c44e57a0a02b/go.mod h1:sb2H21VIVTohCXVrDFo/Skl2tN8Yzba+MXSS6NgCYXw=
-github.com/PeernetOfficial/core/reuseport v0.0.0-20210409140720-c44e57a0a02b h1:9pYGclxvKWSDhe1qlVg8bzeIS78kYehdbNSYautAT+4=
-github.com/PeernetOfficial/core/reuseport v0.0.0-20210409140720-c44e57a0a02b/go.mod h1:j5lvi5jDpdDyuQrsUkCv4kkibVhPWj87sMol41sXIxc=
+github.com/PeernetOfficial/core/dht v0.0.0-20210409204108-27ec231e1475 h1:hnf0sbsHLyX1fjad8XvdBP/wS4mCyzswegIFRbZcjFA=
+github.com/PeernetOfficial/core/dht v0.0.0-20210409204108-27ec231e1475/go.mod h1:sb2H21VIVTohCXVrDFo/Skl2tN8Yzba+MXSS6NgCYXw=
+github.com/PeernetOfficial/core/reuseport v0.0.0-20210409204108-27ec231e1475 h1:e3m7QUra7/QJhv1wTfYp5b1vdwAQuaYMN/oy2oa1faM=
+github.com/PeernetOfficial/core/reuseport v0.0.0-20210409204108-27ec231e1475/go.mod h1:j5lvi5jDpdDyuQrsUkCv4kkibVhPWj87sMol41sXIxc=
 github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=
 github.com/btcsuite/btcd v0.20.1-beta/go.mod h1:wVuoA8VJLEcwgqHBwHmzLRazpKxTv13Px/pDuV7OomQ=
 github.com/btcsuite/btcd v0.21.0-beta.0.20210401013323-36a96f6a0025 h1:aoVqvZk4mLyF3WZbqEVPq+vXnwL2wekZg4P4mjYJNLs=
@@ -46,8 +46,8 @@ golang.org/x/net v0.0.0-20180719180050-a680a1efc54d/go.mod h1:mL1N/T3taQHkDXs73r
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
-golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4 h1:4nGaVu0QrbjT/AK2PRLuQfQuh6DJve+pELhqTdAj3x0=
-golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
+golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1 h1:4qWs8cYYH6PoEFy4dfhDFgoMGkwAcETd+MmPdCPMzUc=
+golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1/go.mod h1:9tjilg8BloeKEkVJvy7fQ90B1CfIiPueXVOjqfkSzI8=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -59,6 +59,7 @@ golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=