feat: start in dev mode (#2347)

cmd/bee/cmd/cmd.go

@@ -71,6 +71,7 @@ const (
 	optionWarmUpTime                     = "warmup-time"
 	optionNameMainNet                    = "mainnet"
 	optionNameRetrievalCaching           = "cache-retrieval"
+	optionNameDevReserveCapacity         = "dev-reserve-capacity"
 )
 
 func init() {
@@ -118,6 +119,10 @@ func newCommand(opts ...option) (c *command, err error) {
 		return nil, err
 	}
 
+	if err := c.initStartDevCmd(); err != nil {
+		return nil, err
+	}
+
 	if err := c.initInitCmd(); err != nil {
 		return nil, err
 	}

cmd/bee/cmd/start_dev.go

+// Copyright 2021 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package cmd
+
+import (
+	"context"
+	"fmt"
+
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/ethersphere/bee/pkg/node"
+	"github.com/kardianos/service"
+	"github.com/spf13/cobra"
+)
+
+func (c *command) initStartDevCmd() (err error) {
+
+	cmd := &cobra.Command{
+		Use:   "dev",
+		Short: "Start a Swarm node in development mode",
+		RunE: func(cmd *cobra.Command, args []string) (err error) {
+			if len(args) > 0 {
+				return cmd.Help()
+			}
+
+			v := strings.ToLower(c.config.GetString(optionNameVerbosity))
+			logger, err := newLogger(cmd, v)
+			if err != nil {
+				return fmt.Errorf("new logger: %v", err)
+			}
+
+			isWindowsService, err := isWindowsService()
+			if err != nil {
+				return fmt.Errorf("failed to determine if we are running in service: %w", err)
+			}
+
+			if isWindowsService {
+				var err error
+				logger, err = createWindowsEventLogger(serviceName, logger)
+				if err != nil {
+					return fmt.Errorf("failed to create windows logger %w", err)
+				}
+			}
+
+			beeASCII := `
+ (                      *        )  (           
+ )\ )                 (  *    ( /(  )\ )        
+(()/(   (    (   (    )\))(   )\())(()/(   (    
+ /(_))  )\   )\  )\  ((_)()\ ((_)\  /(_))  )\   
+(_))_  ((_) ((_)((_) (_()((_)  ((_)(_))_  ((_)  
+ |   \ | __|\ \ / /  |  \/  | / _ \ |   \ | __| 
+ | |) || _|  \ V /   | |\/| || (_) || |) || _|  
+ |___/ |___|  \_/    |_|  |_| \___/ |___/ |___|                
+`
+
+			fmt.Println(beeASCII)
+			fmt.Println()
+			fmt.Println("Starting in development mode")
+			fmt.Println()
+
+			debugAPIAddr := c.config.GetString(optionNameDebugAPIAddr)
+			if !c.config.GetBool(optionNameDebugAPIEnable) {
+				debugAPIAddr = ""
+			}
+
+			// generate signer in here
+			b, err := node.NewDevBee(logger, &node.DevOptions{
+				APIAddr:                  c.config.GetString(optionNameAPIAddr),
+				DebugAPIAddr:             debugAPIAddr,
+				Logger:                   logger,
+				DBOpenFilesLimit:         c.config.GetUint64(optionNameDBOpenFilesLimit),
+				DBBlockCacheCapacity:     c.config.GetUint64(optionNameDBBlockCacheCapacity),
+				DBWriteBufferSize:        c.config.GetUint64(optionNameDBWriteBufferSize),
+				DBDisableSeeksCompaction: c.config.GetBool(optionNameDBDisableSeeksCompaction),
+				CORSAllowedOrigins:       c.config.GetStringSlice(optionCORSAllowedOrigins),
+				ReserveCapacity:          c.config.GetUint64(optionNameDevReserveCapacity),
+			})
+			if err != nil {
+				return err
+			}
+
+			// Wait for termination or interrupt signals.
+			// We want to clean up things at the end.
+			interruptChannel := make(chan os.Signal, 1)
+			signal.Notify(interruptChannel, syscall.SIGINT, syscall.SIGTERM)
+
+			p := &program{
+				start: func() {
+					// Block main goroutine until it is interrupted
+					sig := <-interruptChannel
+
+					logger.Debugf("received signal: %v", sig)
+					logger.Info("shutting down")
+				},
+				stop: func() {
+					// Shutdown
+					done := make(chan struct{})
+					go func() {
+						defer close(done)
+
+						ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+						defer cancel()
+
+						if err := b.Shutdown(ctx); err != nil {
+							logger.Errorf("shutdown: %v", err)
+						}
+					}()
+
+					// If shutdown function is blocking too long,
+					// allow process termination by receiving another signal.
+					select {
+					case sig := <-interruptChannel:
+						logger.Debugf("received signal: %v", sig)
+					case <-done:
+					}
+				},
+			}
+
+			if isWindowsService {
+				s, err := service.New(p, &service.Config{
+					Name:        serviceName,
+					DisplayName: "Bee",
+					Description: "Bee, Swarm client.",
+				})
+				if err != nil {
+					return err
+				}
+
+				if err = s.Run(); err != nil {
+					return err
+				}
+			} else {
+				// start blocks until some interrupt is received
+				p.start()
+				p.stop()
+			}
+
+			return nil
+		},
+		PreRunE: func(cmd *cobra.Command, args []string) error {
+			return c.config.BindPFlags(cmd.Flags())
+		},
+	}
+
+	cmd.Flags().Bool(optionNameDebugAPIEnable, true, "enable debug HTTP API")
+	cmd.Flags().String(optionNameAPIAddr, ":1633", "HTTP API listen address")
+	cmd.Flags().String(optionNameDebugAPIAddr, ":1635", "debug HTTP API listen address")
+	cmd.Flags().String(optionNameVerbosity, "info", "log verbosity level 0=silent, 1=error, 2=warn, 3=info, 4=debug, 5=trace")
+	cmd.Flags().Uint64(optionNameDevReserveCapacity, 4194304, "cache reserve capacity")
+	cmd.Flags().StringSlice(optionCORSAllowedOrigins, []string{}, "origins with CORS headers enabled")
+	cmd.Flags().Uint64(optionNameDBOpenFilesLimit, 200, "number of open files allowed by database")
+	cmd.Flags().Uint64(optionNameDBBlockCacheCapacity, 32*1024*1024, "size of block cache of the database in bytes")
+	cmd.Flags().Uint64(optionNameDBWriteBufferSize, 32*1024*1024, "size of the database write buffer in bytes")
+	cmd.Flags().Bool(optionNameDBDisableSeeksCompaction, false, "disables db compactions triggered by seeks")
+
+	c.root.AddCommand(cmd)
+	return nil
+}

pkg/debugapi/debugapi.go

@@ -14,6 +14,7 @@ import (
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
+
 	"github.com/ethersphere/bee/pkg/accounting"
 	"github.com/ethersphere/bee/pkg/logging"
 	"github.com/ethersphere/bee/pkg/p2p"

pkg/postage/mock/service.go

@@ -7,6 +7,7 @@ package mock
 import (
 	"errors"
 	"math/big"
+	"sync"
 
 	"github.com/ethersphere/bee/pkg/postage"
 )
@@ -22,7 +23,9 @@ func (f optionFunc) apply(r *mockPostage) { f(r) }
 
 // New creates a new mock postage service.
 func New(o ...Option) postage.Service {
-	m := &mockPostage{}
+	m := &mockPostage{
+		issuersMap: make(map[string]*postage.StampIssuer),
+	}
 	for _, v := range o {
 		v.apply(m)
 	}
@@ -37,20 +40,33 @@ func WithAcceptAll() Option {
 }
 
 func WithIssuer(s *postage.StampIssuer) Option {
-	return optionFunc(func(m *mockPostage) { m.i = s })
+	return optionFunc(func(m *mockPostage) {
+		m.issuersMap = map[string]*postage.StampIssuer{string(s.ID()): s}
+	})
 }
 
 type mockPostage struct {
-	i         *postage.StampIssuer
-	acceptAll bool
+	issuersMap map[string]*postage.StampIssuer
+	issuerLock sync.Mutex
+	acceptAll  bool
 }
 
 func (m *mockPostage) Add(s *postage.StampIssuer) {
-	m.i = s
+	m.issuerLock.Lock()
+	defer m.issuerLock.Unlock()
+
+	m.issuersMap[string(s.ID())] = s
 }
 
 func (m *mockPostage) StampIssuers() []*postage.StampIssuer {
-	return []*postage.StampIssuer{m.i}
+	m.issuerLock.Lock()
+	defer m.issuerLock.Unlock()
+
+	issuers := []*postage.StampIssuer{}
+	for _, v := range m.issuersMap {
+		issuers = append(issuers, v)
+	}
+	return issuers
 }
 
 func (m *mockPostage) GetStampIssuer(id []byte) (*postage.StampIssuer, error) {
@@ -58,11 +74,14 @@ func (m *mockPostage) GetStampIssuer(id []byte) (*postage.StampIssuer, error) {
 		return postage.NewStampIssuer("test fallback", "test identity", id, big.NewInt(3), 24, 6, 1000, true), nil
 	}
 
-	if m.i != nil {
-		return m.i, nil
-	}
+	m.issuerLock.Lock()
+	defer m.issuerLock.Unlock()
 
-	return nil, errors.New("stampissuer not found")
+	i, exists := m.issuersMap[string(id)]
+	if !exists {
+		return nil, errors.New("stampissuer not found")
+	}
+	return i, nil
 }
 
 func (m *mockPostage) IssuerUsable(_ *postage.StampIssuer) bool {

pkg/statestore/leveldb/leveldb.go

@@ -14,6 +14,10 @@ import (
 	"github.com/ethersphere/bee/pkg/storage"
 	"github.com/syndtr/goleveldb/leveldb"
 	ldberr "github.com/syndtr/goleveldb/leveldb/errors"
+
+	ldb "github.com/syndtr/goleveldb/leveldb"
+	ldbs "github.com/syndtr/goleveldb/leveldb/storage"
+
 	"github.com/syndtr/goleveldb/leveldb/util"
 )
 
@@ -25,6 +29,24 @@ type store struct {
 	logger logging.Logger
 }
 
+func NewInMemoryStateStore(l logging.Logger) (storage.StateStorer, error) {
+	ldb, err := ldb.Open(ldbs.NewMemStorage(), nil)
+	if err != nil {
+		return nil, err
+	}
+
+	s := &store{
+		db:     ldb,
+		logger: l,
+	}
+
+	if err := migrate(s); err != nil {
+		return nil, err
+	}
+
+	return s, nil
+}
+
 // NewStateStore creates a new persistent state storage.
 func NewStateStore(path string, l logging.Logger) (storage.StateStorer, error) {
 	db, err := leveldb.OpenFile(path, nil)
@@ -46,26 +68,34 @@ func NewStateStore(path string, l logging.Logger) (storage.StateStorer, error) {
 		logger: l,
 	}
 
+	if err := migrate(s); err != nil {
+		return nil, err
+	}
+
+	return s, nil
+}
+
+func migrate(s *store) error {
 	sn, err := s.getSchemaName()
 	if err != nil {
 		if !errors.Is(err, storage.ErrNotFound) {
 			_ = s.Close()
-			return nil, fmt.Errorf("get schema name: %w", err)
+			return fmt.Errorf("get schema name: %w", err)
 		}
 		// new statestore - put schema key with current name
 		if err := s.putSchemaName(dbSchemaCurrent); err != nil {
 			_ = s.Close()
-			return nil, fmt.Errorf("put schema name: %w", err)
+			return fmt.Errorf("put schema name: %w", err)
 		}
 		sn = dbSchemaCurrent
 	}
 
 	if err = s.migrate(sn); err != nil {
 		_ = s.Close()
-		return nil, fmt.Errorf("migrate: %w", err)
+		return fmt.Errorf("migrate: %w", err)
 	}
 
-	return s, nil
+	return nil
 }
 
 // Get retrieves a value of the requested key. If no results are found,

pkg/topology/kademlia/kademlia.go

@@ -182,6 +182,10 @@ func (k *Kad) generateCommonBinPrefixes() {
 		for j := range binPrefixes[i] {
 			pseudoAddrBytes := binPrefixes[i][j].Bytes()
 
+			if len(pseudoAddrBytes) < 1 {
+				continue
+			}
+
 			// flip first bit for bin
 			indexByte, posBit := i/8, i%8
 			if hasBit(bits.Reverse8(pseudoAddrBytes[indexByte]), uint8(posBit)) {

pkg/transaction/mock/transaction.go

@@ -78,7 +78,7 @@ func (m *transactionServiceMock) StoredTransaction(txHash common.Hash) (*transac
 }
 
 func (m *transactionServiceMock) CancelTransaction(ctx context.Context, originalTxHash common.Hash) (common.Hash, error) {
-	if m.send != nil {
+	if m.cancelTransaction != nil {
 		return m.cancelTransaction(ctx, originalTxHash)
 	}
 	return common.Hash{}, errors.New("not implemented")
@@ -133,6 +133,12 @@ func WithResendTransactionFunc(f func(ctx context.Context, txHash common.Hash) e
 	})
 }
 
+func WithCancelTransactionFunc(f func(ctx context.Context, originalTxHash common.Hash) (common.Hash, error)) Option {
+	return optionFunc(func(s *transactionServiceMock) {
+		s.cancelTransaction = f
+	})
+}
+
 func New(opts ...Option) transaction.Service {
 	mock := new(transactionServiceMock)
 	for _, o := range opts {