refactor database access and peer discovery (#4)

* update the db abstraction, and pass context in places

* add a second static identity keys for testing

* keep track of connected peers for dialback later

* widen the number of discovery peers, attempt redial

* update readme

* display log what function the worker is installing

* will rewrite tests

* chore: Updated coverage badge.

Co-authored-by: GitHub Action <[email protected]>

README.md

@@ -1,54 +1,43 @@
-![Coverage](https://img.shields.io/badge/Coverage-48.1%25-yellow)
+![Coverage](https://img.shields.io/badge/Coverage-0.0%25-red)
 
 # b7s daemon
 
-`blockless` the peer to peer networking daemon for the blockless network.
+b7s is a peer-to-peer networking daemon for the blockless network. It is supported on Windows, Linux, and MacOS platforms for both x64 and arm64 architectures.
 
-Supported Platforms
+## Installation
 
-| OS      | arm64 | x64 |
-| ------- | ----- | --- |
-| Windows |       | x   |
-| Linux   | x     | x   |
-| MacOS   | x     | x   |
-
-Using **curl**:
+You can install b7s using either curl or wget:
 
 ```bash
+# using curl
 sudo sh -c "curl https://raw.githubusercontent.com/blocklessnetwork/b7s/main/download.sh | bash"
-```
 
-Using **wget**:
-
-```bash
+# using wget
 sudo sh -c "wget https://raw.githubusercontent.com/blocklessnetwork/b7s/main/download.sh -v -O download.sh; chmod +x download.sh; ./download.sh; rm -rf download.sh"
 ```
 
-Use `docker` see docker [docs](docker/README.md)
+You can also use Docker to install b7s. See the [Docker documentation](https://chat.openai.com/chat/docker/README.md) for more information.
 
-## usage
+Usage
+b7s can be run with a number of commands and flags:
 
-commands
-`b7s [command]`
+Commands:
 
-- `help` display help menu
-- `keygen` generate identity keys for the node
+- `help`: display the help menu
+- `keygen`: generate identity keys for the node
+  Flags:
 
-flags
-`b7s --flag value`
+- `config`: path to the configuration file
+- `out`: style of logging used in the daemon (rich, text, or json)
+  For example:
 
-- `config` path to the configuration file
-- `out` style of logging used in the daemon (rich|text|json)
-
-```bash
-b7s --config=../configs/head-config.yaml --out=json
-```
+## Dependencies
 
-## depends on
+b7s depends on the following repositories:
 
-- [blocklessnetwork/runtime](https://github.com/blocklessnetwork/runtime)
-- [blocklessnetwork/orchestration-chain](https://github.com/blocklessnetwork/orchestration-chain)
+- blocklessnetwork/runtime
+- blocklessnetwork/orchestration-chain
 
-## contributing
+## Contributing
 
-see [src/readme](src/README.md)
+See src/README for information on contributing to the b7s project.

configs/head-config.yaml

@@ -2,7 +2,7 @@ node:
   name: head#1
   ip: 0.0.0.0
   port: 9527
-  key_path: /Users/derekanderson/Projects/blockless/net2/configs/testkeys/priv.bin
+  key_path: /Users/derekanderson/Projects/blockless/net2/configs/testkeys/ident1/priv.bin
   use_static_keys: true
   workspace_root: /tmp/head
   runtime_path: /tmp/runtime

configs/testkeys/ident2/identity

@@ -0,0 +1 @@
+12D3KooWQrN5U3BApv4JYjE5HyKXFKkRF2U8c5FgK3zMPjzkZTpQ

configs/testkeys/ident2/priv.bin

@@ -0,0 +1,2 @@
+
@��!������Y�����F��g��o�����b\�Mɏd�n
+��o}�m���VzwC��5

configs/testkeys/ident2/pub.bin

@@ -0,0 +1,2 @@
+
 �b\�Mɏd�n
+��o}�m���VzwC��5

configs/worker-config.yaml

@@ -3,9 +3,9 @@ node:
   ip: 0.0.0.0
   # 0 means random port
   port: 0
-  # join the blockless network by boot strapping to one of our worker nodes
+  key_path: /Users/derekanderson/Projects/blockless/net2/configs/testkeys/ident2/priv.bin
   boot_nodes:
-    - /ip4/147.75.84.103/tcp/32536/p2p/12D3KooWDCELEqCuHGqqRiMwSjPFLAKnVjL6mGVpSHASmmxTqovc
+    - /ip4/0.0.0.0/tcp/9527/p2p/12D3KooWH9GerdSEroL2nqjpd2GuE5dwmqNi7uHX7FoywBdKcP4q
   workspace_root: /tmp/worker
   runtime_path: /tmp/runtime
 rest:

src/12D3KooWQrN5U3BApv4JYjE5HyKXFKkRF2U8c5FgK3zMPjzkZTpQ_appDb/OPTIONS-000003

@@ -0,0 +1,45 @@
+[Version]
+  pebble_version=0.1
+
+[Options]
+  bytes_per_sync=524288
+  cache_size=8388608
+  cleaner=delete
+  compaction_debt_concurrency=1073741824
+  comparer=leveldb.BytewiseComparator
+  disable_wal=false
+  flush_delay_delete_range=0s
+  flush_delay_range_key=0s
+  flush_split_bytes=4194304
+  format_major_version=1
+  l0_compaction_concurrency=10
+  l0_compaction_file_threshold=500
+  l0_compaction_threshold=4
+  l0_stop_writes_threshold=12
+  lbase_max_bytes=67108864
+  max_concurrent_compactions=1
+  max_manifest_file_size=134217728
+  max_open_files=1000
+  mem_table_size=4194304
+  mem_table_stop_writes_threshold=2
+  min_deletion_rate=0
+  merger=pebble.concatenate
+  read_compaction_rate=16000
+  read_sampling_multiplier=16
+  strict_wal_tail=true
+  table_cache_shards=8
+  table_property_collectors=[]
+  validate_on_ingest=false
+  wal_dir=
+  wal_bytes_per_sync=0
+  max_writer_concurrency=0
+  force_writer_parallelism=false
+
+[Level "0"]
+  block_restart_interval=16
+  block_size=4096
+  compression=Snappy
+  filter_policy=none
+  filter_type=table
+  index_block_size=4096
+  target_file_size=2097152

src/controller/controller.go

@@ -11,14 +11,12 @@ import (
 	"github.com/blocklessnetworking/b7s/src/messaging"
 	"github.com/blocklessnetworking/b7s/src/models"
 	"github.com/blocklessnetworking/b7s/src/repository"
-	"github.com/cockroachdb/pebble"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	log "github.com/sirupsen/logrus"
 )
 
 func IsFunctionInstalled(ctx context.Context, functionId string) (models.FunctionManifest, error) {
-	appDb := ctx.Value("appDb").(*pebble.DB)
-	functionManifestString, err := db.Value(appDb, functionId)
+	functionManifestString, err := db.GetString(ctx, functionId)
 	functionManifest := models.FunctionManifest{}
 
 	json.Unmarshal([]byte(functionManifestString), &functionManifest)
@@ -64,7 +62,7 @@ func MsgInstallFunction(ctx context.Context, installRequest models.RequestFuncti
 		ManifestUrl: manifestPath,
 	}
 
-	log.Debug("request to message peer for install function")
+	log.Info("request to message peer for install function ", msg.ManifestUrl)
 	messaging.PublishMessage(ctx, ctx.Value("topic").(*pubsub.Topic), msg)
 }
 

src/daemon/daemon.go

@@ -26,6 +26,8 @@ import (
 func Run(cmd *cobra.Command, args []string, configPath string) {
 	topicName := "blockless/b7s/general"
 	ctx := context.Background()
+	ctx = context.WithValue(ctx, "topicName", topicName)
+
 	ex, err := os.Executable()
 	if err != nil {
 		log.Warn(err)
@@ -53,13 +55,19 @@ func Run(cmd *cobra.Command, args []string, configPath string) {
 	ctx = setupChannels(ctx)
 
 	// create a new libp2p host
-	host := host.NewHost(ctx, port, config.C.Node.IP)
-	ctx = context.WithValue(ctx, "host", host)
+	h := host.NewHost(ctx, port, config.C.Node.IP)
+	ctx = context.WithValue(ctx, "host", h)
 
 	// set appdb config
-	appDb := db.Get(exPath + "/" + host.ID().Pretty() + "_appDb")
+	appDb := db.GetDb(exPath + "/" + h.ID().Pretty() + "_appDb")
 	ctx = context.WithValue(ctx, "appDb", appDb)
 
+	n := &host.ConnectedNotifee{
+		Ctx: ctx,
+	}
+
+	h.Network().Notify(n)
+
 	// response memstore
 	// todo flush memstore occasionally
 	executionResponseMemStore := memstore.NewReqRespStore()
@@ -70,7 +78,7 @@ func Run(cmd *cobra.Command, args []string, configPath string) {
 	go listenToChannels(ctx)
 
 	// pubsub topics from p2p
-	topic := messaging.Subscribe(ctx, host, topicName)
+	topic := messaging.Subscribe(ctx, h, topicName)
 	ctx = context.WithValue(ctx, "topic", topic)
 
 	// start health monitoring
@@ -86,7 +94,7 @@ func Run(cmd *cobra.Command, args []string, configPath string) {
 	defer ticker.Stop()
 
 	// discover peers
-	go dht.DiscoverPeers(ctx, host, topicName)
+	go dht.DiscoverPeers(ctx, h)
 
 	// daemon is running
 	// waiting for ctrl-c to exit

src/db/db.go

@@ -1,31 +1,44 @@
 package db
 
 import (
+	"context"
+
 	log "github.com/sirupsen/logrus"
 
 	"github.com/cockroachdb/pebble"
 )
 
-func Get(DatabaseId string) *pebble.DB {
-
+func GetDb(DatabaseId string) *pebble.DB {
 	dbPath := DatabaseId
-	log.Info("Opening database: ", dbPath)
+	log.Info("opening database: ", dbPath)
 	db, err := pebble.Open(dbPath, &pebble.Options{})
 	if err != nil {
 		log.Warn(err)
 	}
 	return db
 }
 
-func Set(db *pebble.DB, key string, value string) error {
+func Set(ctx context.Context, key string, value string) error {
+	db := ctx.Value("appDb").(*pebble.DB)
 	if err := db.Set([]byte(key), []byte(value), pebble.Sync); err != nil {
 		log.Warn(err)
 		return err
 	}
 	return nil
 }
 
-func Value(db *pebble.DB, key string) (string, error) {
+func Get(ctx context.Context, key string) ([]byte, error) {
+	db := ctx.Value("appDb").(*pebble.DB)
+	value, closer, err := db.Get([]byte(key))
+	if err != nil {
+		return nil, err
+	}
+	defer closer.Close()
+	return value, nil
+}
+
+func GetString(ctx context.Context, key string) (string, error) {
+	db := ctx.Value("appDb").(*pebble.DB)
 	value, closer, err := db.Get([]byte(key))
 	if err != nil {
 		return "", err