func (db *DB) po(addr swarm.Address) (bin uint8) {
	return swarm.Proximity(db.baseKey, addr.Bytes())
}
po computes the proximity order between the address
and database base key.
package localstore
import (
	"encoding/binary"
	"errors"
	"os"
	"runtime/pprof"
	"sync"
	"time"
	"github.com/ethsana/sana/pkg/logging"
	"github.com/ethsana/sana/pkg/pinning"
	"github.com/ethsana/sana/pkg/postage"
	"github.com/ethsana/sana/pkg/postage/batchstore"
	"github.com/ethsana/sana/pkg/shed"
	"github.com/ethsana/sana/pkg/storage"
	"github.com/ethsana/sana/pkg/swarm"
	"github.com/ethsana/sana/pkg/tags"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/syndtr/goleveldb/leveldb"
)
var _ storage.Storer = &DB{}
var (
	ErrInvalidMode = errors.New("invalid mode")
)
var (
	defaultCacheCapacity uint64 = 1000000
	maxParallelUpdateGC = 1000
	flipFlopBufferDuration    = 150 * time.Millisecond
	flipFlopWorstCaseDuration = 20 * time.Second
)
type DB struct {
	shed *shed.DB
	tags *tags.Tags
	stateStore storage.StateStorer
	schemaName shed.StringField
	retrievalDataIndex   shed.Index
	retrievalAccessIndex shed.Index
	pushIndex shed.Index
	pushTriggers   []chan<- struct{}
	pushTriggersMu sync.RWMutex
	pullIndex shed.Index
	pullTriggers   map[uint8][]chan<- struct{}
	pullTriggersMu sync.RWMutex
	binIDs shed.Uint64Vector
	gcIndex shed.Index
	pinIndex shed.Index
	postageChunksIndex shed.Index
	postageRadiusIndex shed.Index
	postageIndexIndex shed.Index
	gcSize shed.Uint64Field
	reserveSize shed.Uint64Field
	cacheCapacity uint64
	reserveCapacity uint64
	unreserveFunc func(postage.UnreserveIteratorFn) error
	collectGarbageTrigger chan struct{}
	reserveEvictionTrigger chan struct{}
	updateGCSem chan struct{}
	updateGCWG sync.WaitGroup
	baseKey []byte
	batchMu sync.Mutex
	gcRunning bool
	dirtyAddresses []swarm.Address
	close chan struct{}
	collectGarbageWorkerDone  chan struct{}
	reserveEvictionWorkerDone chan struct{}
	subscritionsWG sync.WaitGroup
	metrics metrics
	logger logging.Logger
}
type Options struct {
	Capacity uint64
	ReserveCapacity uint64
	UnreserveFunc func(postage.UnreserveIteratorFn) error
	OpenFilesLimit uint64
	BlockCacheCapacity uint64
	WriteBufferSize uint64
	DisableSeeksCompaction bool
	MetricsPrefix string
	Tags          *tags.Tags
}
func New(path string, baseKey []byte, ss storage.StateStorer, o *Options, logger logging.Logger) (db *DB, err error) {
	if o == nil {
		o = &Options{
			Capacity:        defaultCacheCapacity,
			ReserveCapacity: uint64(batchstore.Capacity),
		}
	}
	db = &DB{
		stateStore:      ss,
		cacheCapacity:   o.Capacity,
		reserveCapacity: o.ReserveCapacity,
		unreserveFunc:   o.UnreserveFunc,
		baseKey:         baseKey,
		tags:            o.Tags,
		collectGarbageTrigger:     make(chan struct{}, 1),
		reserveEvictionTrigger:    make(chan struct{}, 1),
		close:                     make(chan struct{}),
		collectGarbageWorkerDone:  make(chan struct{}),
		reserveEvictionWorkerDone: make(chan struct{}),
		metrics:                   newMetrics(),
		logger:                    logger,
	}
	if db.cacheCapacity == 0 {
		db.cacheCapacity = defaultCacheCapacity
	}
	capacityMB := float64((db.cacheCapacity+uint64(batchstore.Capacity))*swarm.ChunkSize) * 9.5367431640625e-7
	if capacityMB <= 1000 {
		db.logger.Infof("database capacity: %d chunks (approximately %fMB)", db.cacheCapacity, capacityMB)
	} else {
		db.logger.Infof("database capacity: %d chunks (approximately %0.1fGB)", db.cacheCapacity, capacityMB/1000)
	}
	if maxParallelUpdateGC > 0 {
		db.updateGCSem = make(chan struct{}, maxParallelUpdateGC)
	}
	shedOpts := &shed.Options{
		OpenFilesLimit:         o.OpenFilesLimit,
		BlockCacheCapacity:     o.BlockCacheCapacity,
		WriteBufferSize:        o.WriteBufferSize,
		DisableSeeksCompaction: o.DisableSeeksCompaction,
	}
	if withinRadiusFn == nil {
		withinRadiusFn = withinRadius
	}
	db.shed, err = shed.NewDB(path, shedOpts)
	if err != nil {
		return nil, err
	}
	db.schemaName, err = db.shed.NewStringField("schema-name")
	if err != nil {
		return nil, err
	}
	schemaName, err := db.schemaName.Get()
	if err != nil && !errors.Is(err, leveldb.ErrNotFound) {
		return nil, err
	}
	if schemaName == "" {
		err := db.schemaName.Put(DBSchemaCurrent)
		if err != nil {
			return nil, err
		}
	} else {
		err = db.migrate(schemaName)
		if err != nil {
			return nil, err
		}
	}
	db.gcSize, err = db.shed.NewUint64Field("gc-size")
	if err != nil {
		return nil, err
	}
	db.reserveSize, err = db.shed.NewUint64Field("reserve-size")
	if err != nil {
		return nil, err
	}
	headerSize := 16 + postage.StampSize
	db.retrievalDataIndex, err = db.shed.NewIndex("Address->StoreTimestamp|BinID|BatchID|BatchIndex|Sig|Data", shed.IndexFuncs{
		EncodeKey: func(fields shed.Item) (key []byte, err error) {
			return fields.Address, nil
		},
		DecodeKey: func(key []byte) (e shed.Item, err error) {
			e.Address = key
			return e, nil
		},
		EncodeValue: func(fields shed.Item) (value []byte, err error) {
			b := make([]byte, headerSize)
			binary.BigEndian.PutUint64(b[:8], fields.BinID)
			binary.BigEndian.PutUint64(b[8:16], uint64(fields.StoreTimestamp))
			stamp, err := postage.NewStamp(fields.BatchID, fields.Index, fields.Timestamp, fields.Sig).MarshalBinary()
			if err != nil {
				return nil, err
			}
			copy(b[16:], stamp)
			value = append(b, fields.Data...)
			return value, nil
		},
		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
			e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[8:16]))
			e.BinID = binary.BigEndian.Uint64(value[:8])
			stamp := new(postage.Stamp)
			if err = stamp.UnmarshalBinary(value[16:headerSize]); err != nil {
				return e, err
			}
			e.BatchID = stamp.BatchID()
			e.Index = stamp.Index()
			e.Timestamp = stamp.Timestamp()
			e.Sig = stamp.Sig()
			e.Data = value[headerSize:]
			return e, nil
		},
	})
	if err != nil {
		return nil, err
	}
	db.retrievalAccessIndex, err = db.shed.NewIndex("Address->AccessTimestamp", shed.IndexFuncs{
		EncodeKey: func(fields shed.Item) (key []byte, err error) {
			return fields.Address, nil
		},
		DecodeKey: func(key []byte) (e shed.Item, err error) {
			e.Address = key
			return e, nil
		},
		EncodeValue: func(fields shed.Item) (value []byte, err error) {
			b := make([]byte, 8)
			binary.BigEndian.PutUint64(b, uint64(fields.AccessTimestamp))
			return b, nil
		},
		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
			e.AccessTimestamp = int64(binary.BigEndian.Uint64(value))
			return e, nil
		},
	})
	if err != nil {
		return nil, err
	}
	db.pullIndex, err = db.shed.NewIndex("PO|BinID->Hash", shed.IndexFuncs{
		EncodeKey: func(fields shed.Item) (key []byte, err error) {
			key = make([]byte, 9)
			key[0] = db.po(swarm.NewAddress(fields.Address))
			binary.BigEndian.PutUint64(key[1:9], fields.BinID)
			return key, nil
		},
		DecodeKey: func(key []byte) (e shed.Item, err error) {
			e.BinID = binary.BigEndian.Uint64(key[1:9])
			return e, nil
		},
		EncodeValue: func(fields shed.Item) (value []byte, err error) {
			value = make([]byte, 64)  
			copy(value, fields.Address)
			copy(value[32:], fields.BatchID)
			return value, nil
		},
		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
			e.Address = value[:32]
			e.BatchID = value[32:64]
			return e, nil
		},
	})
	if err != nil {
		return nil, err
	}
	db.binIDs, err = db.shed.NewUint64Vector("bin-ids")
	if err != nil {
		return nil, err
	}
	db.pullTriggers = make(map[uint8][]chan<- struct{})
	db.pushIndex, err = db.shed.NewIndex("StoreTimestamp|Hash->Tags", shed.IndexFuncs{
		EncodeKey: func(fields shed.Item) (key []byte, err error) {
			key = make([]byte, 40)
			binary.BigEndian.PutUint64(key[:8], uint64(fields.StoreTimestamp))
			copy(key[8:], fields.Address)
			return key, nil
		},
		DecodeKey: func(key []byte) (e shed.Item, err error) {
			e.Address = key[8:]
			e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[:8]))
			return e, nil
		},
		EncodeValue: func(fields shed.Item) (value []byte, err error) {
			tag := make([]byte, 4)
			binary.BigEndian.PutUint32(tag, fields.Tag)
			return tag, nil
		},
		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
			if len(value) == 4 {  
				e.Tag = binary.BigEndian.Uint32(value)
			}
			return e, nil
		},
	})
	if err != nil {
		return nil, err
	}
	db.pushTriggers = make([]chan<- struct{}, 0)
	db.gcIndex, err = db.shed.NewIndex("AccessTimestamp|BinID|Hash->BatchID|BatchIndex", shed.IndexFuncs{
		EncodeKey: func(fields shed.Item) (key []byte, err error) {
			b := make([]byte, 16, 16+len(fields.Address))
			binary.BigEndian.PutUint64(b[:8], uint64(fields.AccessTimestamp))
			binary.BigEndian.PutUint64(b[8:16], fields.BinID)
			key = append(b, fields.Address...)
			return key, nil
		},
		DecodeKey: func(key []byte) (e shed.Item, err error) {
			e.AccessTimestamp = int64(binary.BigEndian.Uint64(key[:8]))
			e.BinID = binary.BigEndian.Uint64(key[8:16])
			e.Address = key[16:]
			return e, nil
		},
		EncodeValue: func(fields shed.Item) (value []byte, err error) {
			value = make([]byte, 40)
			copy(value, fields.BatchID)
			copy(value[32:], fields.Index)
			return value, nil
		},
		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
			e.BatchID = make([]byte, 32)
			copy(e.BatchID, value[:32])
			e.Index = make([]byte, postage.IndexSize)
			copy(e.Index, value[32:])
			return e, nil
		},
	})
	if err != nil {
		return nil, err
	}
	db.pinIndex, err = db.shed.NewIndex("Hash->PinCounter", shed.IndexFuncs{
		EncodeKey: func(fields shed.Item) (key []byte, err error) {
			return fields.Address, nil
		},
		DecodeKey: func(key []byte) (e shed.Item, err error) {
			e.Address = key
			return e, nil
		},
		EncodeValue: func(fields shed.Item) (value []byte, err error) {
			b := make([]byte, 8)
			binary.BigEndian.PutUint64(b[:8], fields.PinCounter)
			return b, nil
		},
		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
			e.PinCounter = binary.BigEndian.Uint64(value[:8])
			return e, nil
		},
	})
	if err != nil {
		return nil, err
	}
	db.postageChunksIndex, err = db.shed.NewIndex("BatchID|PO|Hash->nil", shed.IndexFuncs{
		EncodeKey: func(fields shed.Item) (key []byte, err error) {
			key = make([]byte, 65)
			copy(key[:32], fields.BatchID)
			key[32] = db.po(swarm.NewAddress(fields.Address))
			copy(key[33:], fields.Address)
			return key, nil
		},
		DecodeKey: func(key []byte) (e shed.Item, err error) {
			e.BatchID = key[:32]
			e.Address = key[33:65]
			return e, nil
		},
		EncodeValue: func(fields shed.Item) (value []byte, err error) {
			return nil, nil
		},
		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
			return e, nil
		},
	})
	if err != nil {
		return nil, err
	}
	db.postageRadiusIndex, err = db.shed.NewIndex("BatchID->Radius", shed.IndexFuncs{
		EncodeKey: func(fields shed.Item) (key []byte, err error) {
			key = make([]byte, 32)
			copy(key[:32], fields.BatchID)
			return key, nil
		},
		DecodeKey: func(key []byte) (e shed.Item, err error) {
			e.BatchID = key[:32]
			return e, nil
		},
		EncodeValue: func(fields shed.Item) (value []byte, err error) {
			return []byte{fields.Radius}, nil
		},
		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
			e.Radius = value[0]
			return e, nil
		},
	})
	if err != nil {
		return nil, err
	}
	db.postageIndexIndex, err = db.shed.NewIndex("BatchID|BatchIndex->Hash|Timestamp", shed.IndexFuncs{
		EncodeKey: func(fields shed.Item) (key []byte, err error) {
			key = make([]byte, 40)
			copy(key[:32], fields.BatchID)
			copy(key[32:40], fields.Index)
			return key, nil
		},
		DecodeKey: func(key []byte) (e shed.Item, err error) {
			e.BatchID = key[:32]
			e.Index = key[32:40]
			return e, nil
		},
		EncodeValue: func(fields shed.Item) (value []byte, err error) {
			value = make([]byte, 40)
			copy(value, fields.Address)
			copy(value[32:], fields.Timestamp)
			return value, nil
		},
		DecodeValue: func(keyItem shed.Item, value []byte) (e shed.Item, err error) {
			e.Address = value[:32]
			e.Timestamp = value[32:]
			return e, nil
		},
	})
	if err != nil {
		return nil, err
	}
	go db.collectGarbageWorker()
	go db.reserveEvictionWorker()
	return db, nil
}
func (db *DB) Close() (err error) {
	close(db.close)
	done := make(chan struct{})
	go func() {
		db.updateGCWG.Wait()
		db.subscritionsWG.Wait()
		<-db.collectGarbageWorkerDone
		<-db.reserveEvictionWorkerDone
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(5 * time.Second):
		db.logger.Errorf("localstore closed with still active goroutines")
		prof := pprof.Lookup("goroutine")
		err = prof.WriteTo(os.Stdout, 2)
		if err != nil {
			return err
		}
	}
	return db.shed.Close()
}