Concurrency testing (#128)

basic concurrency testing

concurrency-test-docker-compose.yml

@@ -0,0 +1,49 @@
+services:
+  diddht-a:
+    build:
+      context: impl
+      dockerfile: build/Dockerfile
+    environment:
+      STORAGE_URI: postgres://postgres:aa@postgres/postgres
+    depends_on:
+      - postgres
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:8305/health"]
+      interval: 5s
+      timeout: 1s
+      retries: 10
+  diddht-b:
+    build:
+      context: impl
+      dockerfile: build/Dockerfile
+    environment:
+      STORAGE_URI: postgres://postgres:aa@postgres/postgres
+    depends_on:
+      - postgres
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:8305/health"]
+      interval: 5s
+      timeout: 1s
+      retries: 10
+  postgres:
+    image: library/postgres
+    command: ["-E"]
+    environment:
+      POSTGRES_PASSWORD: aa
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U postgres"]
+      interval: 5s
+      timeout: 5s
+      retries: 5
+  concurrencytest:
+    build:
+      context: impl
+      dockerfile_inline: |
+        FROM library/golang:latest
+        COPY . /go/diddht
+        WORKDIR /go/diddht
+        RUN go build -o /concurrencytest ./concurrencytest
+    command: "/concurrencytest"
+    depends_on:
+      - diddht-a
+      - diddht-b

impl/concurrencytest/main.go

@@ -0,0 +1,131 @@
+package main
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/TBD54566975/did-dht-method/internal/did"
+	"github.com/TBD54566975/did-dht-method/pkg/dht"
+	"github.com/sirupsen/logrus"
+)
+
+var (
+	iterationsPerServer = 1000
+	servers             = []string{"diddht-a", "diddht-b"}
+)
+
+func main() {
+	logrus.SetLevel(logrus.DebugLevel)
+
+	programstart := time.Now()
+
+	var wg sync.WaitGroup
+	for _, server := range servers {
+		for i := 0; i < iterationsPerServer; i++ {
+			log := logrus.WithField("server", server).WithField("i", i)
+
+			s := server
+			wg.Add(1)
+			go func() {
+				putStart := time.Now()
+				suffix, err := put(s)
+				if err != nil {
+					log = log.WithError(err)
+				}
+				log.WithField("time", time.Since(putStart)).Info("PUT request completed")
+				if err != nil {
+					return
+				}
+
+				getStart := time.Now()
+				if err := get(s, suffix); err != nil {
+					log = log.WithError(err)
+				}
+				log.WithField("time", time.Since(getStart)).Info("GET request completed")
+
+				wg.Done()
+			}()
+		}
+	}
+
+	wg.Wait()
+
+	logrus.WithField("time", time.Since(programstart)).Info("concurrency test completed")
+}
+
+func put(server string) (string, error) {
+	didID, reqData, err := generateDIDPutRequest()
+	if err != nil {
+		return "", err
+	}
+
+	suffix, err := did.DHT(didID).Suffix()
+	if err != nil {
+		return "", err
+	}
+
+	req, err := http.NewRequest(http.MethodPut, "http://"+server+":8305/"+suffix, bytes.NewReader(reqData))
+	if err != nil {
+		return "", err
+	}
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+
+	if resp.StatusCode != 200 {
+		return "", fmt.Errorf("unexpected %s: %s", resp.Status, string(body))
+	}
+
+	return suffix, nil
+}
+
+func get(server, suffix string) error {
+	resp, err := http.Get("http://" + server + ":8305/" + suffix)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	_, err = io.ReadAll(resp.Body)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func generateDIDPutRequest() (string, []byte, error) {
+	// generate a DID Document
+	sk, doc, err := did.GenerateDIDDHT(did.CreateDIDDHTOpts{})
+	if err != nil {
+		return "", nil, err
+	}
+
+	packet, err := did.DHT(doc.ID).ToDNSPacket(*doc, nil)
+	if err != nil {
+		return "", nil, err
+	}
+
+	bep44Put, err := dht.CreatePKARRPublishRequest(sk, *packet)
+	if err != nil {
+		return "", nil, err
+	}
+
+	// prepare request as sig:seq:v
+	var seqBuf [8]byte
+	binary.BigEndian.PutUint64(seqBuf[:], uint64(bep44Put.Seq))
+	return doc.ID, append(bep44Put.Sig[:], append(seqBuf[:], bep44Put.V.([]byte)...)...), nil
+}

impl/config/config.go

@@ -113,9 +113,10 @@ func LoadConfig(path string) (*Config, error) {
 		}
 	}
 
-	if err = applyEnvVariables(cfg); err != nil {
+	if err = applyEnvVariables(&cfg); err != nil {
 		return nil, errors.Wrap(err, "apply env variables")
 	}
+
 	return &cfg, nil
 }
 
@@ -139,7 +140,7 @@ func loadTOMLConfig(path string, cfg *Config) error {
 	return nil
 }
 
-func applyEnvVariables(cfg Config) error {
+func applyEnvVariables(cfg *Config) error {
 	if err := godotenv.Load(DefaultEnvPath); err != nil {
 		// The error indicates that the file or directory does not exist.
 		if os.IsNotExist(err) {
@@ -153,6 +154,12 @@ func applyEnvVariables(cfg Config) error {
 	if present {
 		cfg.DHTConfig.BootstrapPeers = strings.Split(bootstrapPeers, ",")
 	}
+
+	storage, present := os.LookupEnv("STORAGE_URI")
+	if present {
+		cfg.ServerConfig.StorageURI = storage
+	}
+
 	return nil
 }
 

impl/pkg/storage/db/bolt/bolt.go

@@ -5,10 +5,11 @@ import (
 	"encoding/json"
 	"time"
 
-	"github.com/TBD54566975/did-dht-method/pkg/pkarr"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	bolt "go.etcd.io/bbolt"
+
+	"github.com/TBD54566975/did-dht-method/pkg/pkarr"
 )
 
 const (

impl/pkg/storage/db/postgres/postgres.go

@@ -6,11 +6,12 @@ import (
 	"embed"
 	"fmt"
 
-	"github.com/TBD54566975/did-dht-method/pkg/pkarr"
 	pgx "github.com/jackc/pgx/v5"
 	_ "github.com/jackc/pgx/v5/stdlib"
 	goose "github.com/pressly/goose/v3"
 	"github.com/sirupsen/logrus"
+
+	"github.com/TBD54566975/did-dht-method/pkg/pkarr"
 )
 
 //go:embed migrations

impl/pkg/storage/storage.go

@@ -4,6 +4,9 @@ import (
 	"context"
 	"fmt"
 	"net/url"
+	"strings"
+
+	"github.com/sirupsen/logrus"
 
 	"github.com/TBD54566975/did-dht-method/pkg/pkarr"
 	"github.com/TBD54566975/did-dht-method/pkg/storage/db/bolt"
@@ -29,8 +32,13 @@ func NewStorage(uri string) (Storage, error) {
 		if u.Path != "" {
 			filename = fmt.Sprintf("%s/%s", filename, u.Path)
 		}
+		logrus.WithField("file", filename).Info("using boltdb for storage")
 		return bolt.NewBolt(filename)
 	case "postgres":
+		logrus.WithFields(logrus.Fields{
+			"host":     u.Host,
+			"database": strings.TrimPrefix(u.Path, "/"),
+		}).Info("using postgres for storage")
 		return postgres.NewPostgres(uri)
 	default:
 		return nil, fmt.Errorf("unsupported db type %s (from uri %s)", u.Scheme, uri)