diff --git a/swarm/storage/localstore/accessor.go b/swarm/storage/localstore/accessor.go
new file mode 100644
index 0000000000..9cba713d08
--- /dev/null
+++ b/swarm/storage/localstore/accessor.go
@@ -0,0 +1,74 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package localstore
+
+import (
+ "context"
+
+ "github.com/syndtr/goleveldb/leveldb"
+
+ "github.com/ethereum/go-ethereum/swarm/shed"
+ "github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+// Accessor implements ChunkStore to manage data
+// in DB with different modes of access and update.
+type Accessor struct {
+ db *DB
+ mode Mode
+}
+
+// Accessor returns a new Accessor with a specified Mode.
+func (db *DB) Accessor(mode Mode) *Accessor {
+ return &Accessor{
+ mode: mode,
+ db: db,
+ }
+}
+
+// Put overwrites the underlying DB Put method for the specific mode of update.
+func (u *Accessor) Put(ctx context.Context, ch storage.Chunk) error {
+ return u.db.update(ctx, u.mode, chunkToItem(ch))
+}
+
+// Get overwrites the underlying DB Get method for the specific mode of access.
+func (u *Accessor) Get(_ context.Context, addr storage.Address) (chunk storage.Chunk, err error) {
+ item := addressToItem(addr)
+ out, err := u.db.access(u.mode, item)
+ if err != nil {
+ if err == leveldb.ErrNotFound {
+ return nil, storage.ErrChunkNotFound
+ }
+ return nil, err
+ }
+ return storage.NewChunk(out.Address, out.Data), nil
+}
+
+// chunkToItem creates new IndexItem with data provided by the Chunk.
+func chunkToItem(ch storage.Chunk) shed.IndexItem {
+ return shed.IndexItem{
+ Address: ch.Address(),
+ Data: ch.Data(),
+ }
+}
+
+// addressToItem creates new IndexItem with a provided address.
+func addressToItem(addr storage.Address) shed.IndexItem {
+ return shed.IndexItem{
+ Address: addr,
+ }
+}
diff --git a/swarm/storage/localstore/accessor_test.go b/swarm/storage/localstore/accessor_test.go
new file mode 100644
index 0000000000..20a38c0445
--- /dev/null
+++ b/swarm/storage/localstore/accessor_test.go
@@ -0,0 +1,98 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package localstore
+
+import (
+ "bytes"
+ "context"
+ "testing"
+
+ "github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+// TestAccessors tests most basic Put and Get functionalities
+// for different accessors. This test validates that the chunk
+// is retrievable from the database, not if all indexes are set
+// correctly.
+func TestAccessors(t *testing.T) {
+ db, cleanupFunc := newTestDB(t)
+ defer cleanupFunc()
+
+ for _, m := range []Mode{
+ ModeSyncing,
+ ModeUpload,
+ ModeRequest,
+ ModeSynced,
+ ModeAccess,
+ } {
+ t.Run(ModeName(m), func(t *testing.T) {
+ a := db.Accessor(m)
+
+ want := generateRandomChunk()
+
+ err := a.Put(context.Background(), want)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ got, err := a.Get(context.Background(), want.Address())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Equal(got.Data(), want.Data()) {
+ t.Errorf("got chunk data %x, want %x", got.Data(), want.Data())
+ }
+ })
+ }
+
+ // Removal mode is a special case as it removes the chunk
+ // from the database.
+ t.Run(ModeName(ModeRemoval), func(t *testing.T) {
+ a := db.Accessor(ModeUpload)
+
+ want := generateRandomChunk()
+
+ // first put a random chunk to the database
+ err := a.Put(context.Background(), want)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ got, err := a.Get(context.Background(), want.Address())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Equal(got.Data(), want.Data()) {
+ t.Errorf("got chunk data %x, want %x", got.Data(), want.Data())
+ }
+
+ a = db.Accessor(ModeRemoval)
+
+ // removal accessor actually removes the chunk on Put
+ err = a.Put(context.Background(), want)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ // chunk should not be found
+ wantErr := storage.ErrChunkNotFound
+ _, err = a.Get(context.Background(), want.Address())
+ if err != wantErr {
+ t.Errorf("got error %v, expected %v", err, wantErr)
+ }
+ })
+}
diff --git a/swarm/storage/localstore/localstore.go b/swarm/storage/localstore/localstore.go
new file mode 100644
index 0000000000..9c455507bf
--- /dev/null
+++ b/swarm/storage/localstore/localstore.go
@@ -0,0 +1,244 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package localstore
+
+import (
+ "encoding/binary"
+ "errors"
+ "sync"
+ "time"
+
+ "github.com/ethereum/go-ethereum/swarm/shed"
+ "github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+const (
+ // maximal time for DB.Close must return
+ closeTimeout = 10 * time.Second
+)
+
+var (
+ // ErrInvalidMode is retuned when an unkonw Mode
+ // is provided to the function.
+ ErrInvalidMode = errors.New("invalid mode")
+ // ErrDBClosed is returned when database is closed.
+ ErrDBClosed = errors.New("db closed")
+)
+
+// DB is the local store implementation and holds
+// database related objects.
+type DB struct {
+ shed *shed.DB
+
+ // fields and indexes
+ schemaName shed.StringField
+ sizeCounter shed.Uint64Field
+ retrievalIndex shed.Index
+ pushIndex shed.Index
+ pullIndex shed.Index
+ gcIndex shed.Index
+
+ baseKey []byte
+
+ batch *batch // current batch
+ mu sync.RWMutex // mutex for accessing current batch
+ writeTrigger chan struct{} // channel to signal current write batch
+ writeDone chan struct{} // closed when writeBatches function returns
+ close chan struct{} // closed on Close, signals other goroutines to terminate
+}
+
+// New returns a new DB. All fields and indexes are initialized
+// and possible conflicts with schema from existing database is checked.
+// One goroutine for writing batches is created.
+func New(path string, baseKey []byte) (db *DB, err error) {
+ db = &DB{
+ baseKey: baseKey,
+ batch: newBatch(),
+ writeTrigger: make(chan struct{}, 1),
+ close: make(chan struct{}),
+ writeDone: make(chan struct{}),
+ }
+ db.shed, err = shed.NewDB(path)
+ if err != nil {
+ return nil, err
+ }
+ // Identify current storage schema by arbitrary name.
+ db.schemaName, err = db.shed.NewStringField("schema-name")
+ if err != nil {
+ return nil, err
+ }
+ db.sizeCounter, err = db.shed.NewUint64Field("size")
+ if err != nil {
+ return nil, err
+ }
+ db.retrievalIndex, err = db.shed.NewIndex("Hash->StoredTimestamp|AccessTimestamp|Data", shed.IndexFuncs{
+ EncodeKey: func(fields shed.IndexItem) (key []byte, err error) {
+ return fields.Address, nil
+ },
+ DecodeKey: func(key []byte) (e shed.IndexItem, err error) {
+ e.Address = key
+ return e, nil
+ },
+ EncodeValue: func(fields shed.IndexItem) (value []byte, err error) {
+ b := make([]byte, 16)
+ binary.BigEndian.PutUint64(b[:8], uint64(fields.StoreTimestamp))
+ binary.BigEndian.PutUint64(b[8:16], uint64(fields.AccessTimestamp))
+ value = append(b, fields.Data...)
+ return value, nil
+ },
+ DecodeValue: func(value []byte) (e shed.IndexItem, err error) {
+ e.StoreTimestamp = int64(binary.BigEndian.Uint64(value[:8]))
+ e.AccessTimestamp = int64(binary.BigEndian.Uint64(value[8:16]))
+ e.Data = value[16:]
+ return e, nil
+ },
+ })
+ if err != nil {
+ return nil, err
+ }
+ // pull index allows history and live syncing per po bin
+ db.pullIndex, err = db.shed.NewIndex("PO|StoredTimestamp|Hash->nil", shed.IndexFuncs{
+ EncodeKey: func(fields shed.IndexItem) (key []byte, err error) {
+ key = make([]byte, 41)
+ key[0] = db.po(fields.Address)
+ binary.BigEndian.PutUint64(key[1:9], uint64(fields.StoreTimestamp))
+ copy(key[9:], fields.Address[:])
+ return key, nil
+ },
+ DecodeKey: func(key []byte) (e shed.IndexItem, err error) {
+ e.Address = key[9:]
+ e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[1:9]))
+ return e, nil
+ },
+ EncodeValue: func(fields shed.IndexItem) (value []byte, err error) {
+ return nil, nil
+ },
+ DecodeValue: func(value []byte) (e shed.IndexItem, err error) {
+ return e, nil
+ },
+ })
+ if err != nil {
+ return nil, err
+ }
+ // push index contains as yet unsynced chunks
+ db.pushIndex, err = db.shed.NewIndex("StoredTimestamp|Hash->nil", shed.IndexFuncs{
+ EncodeKey: func(fields shed.IndexItem) (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.IndexItem, err error) {
+ e.Address = key[8:]
+ e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[:8]))
+ return e, nil
+ },
+ EncodeValue: func(fields shed.IndexItem) (value []byte, err error) {
+ return nil, nil
+ },
+ DecodeValue: func(value []byte) (e shed.IndexItem, err error) {
+ e.AccessTimestamp = int64(binary.BigEndian.Uint64(value))
+ return e, nil
+ },
+ })
+ if err != nil {
+ return nil, err
+ }
+ // gc index for removable chunk ordered by ascending last access time
+ db.gcIndex, err = db.shed.NewIndex("AccessTimestamp|StoredTimestamp|Hash->nil", shed.IndexFuncs{
+ EncodeKey: func(fields shed.IndexItem) (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], uint64(fields.StoreTimestamp))
+ key = append(b, fields.Address...)
+ return key, nil
+ },
+ DecodeKey: func(key []byte) (e shed.IndexItem, err error) {
+ e.AccessTimestamp = int64(binary.BigEndian.Uint64(key[:8]))
+ e.StoreTimestamp = int64(binary.BigEndian.Uint64(key[8:16]))
+ e.Address = key[16:]
+ return e, nil
+ },
+ EncodeValue: func(fields shed.IndexItem) (value []byte, err error) {
+ return nil, nil
+ },
+ DecodeValue: func(value []byte) (e shed.IndexItem, err error) {
+ return e, nil
+ },
+ })
+ if err != nil {
+ return nil, err
+ }
+ // start goroutine what writes batches
+ go db.writeBatches()
+ return db, nil
+}
+
+// Close closes the underlying database.
+func (db *DB) Close() (err error) {
+ // signal other goroutines that
+ // the database is closing
+ close(db.close)
+ select {
+ // wait for writeBatches to write
+ // the last batch
+ case <-db.writeDone:
+ // closing timeout
+ case <-time.After(closeTimeout):
+ }
+ return db.shed.Close()
+}
+
+// writeBatches is a forever loop handing out the current batch apply
+// the batch when the db is free.
+func (db *DB) writeBatches() {
+ // close the writeDone channel
+ // so the DB.Close can return
+ defer close(db.writeDone)
+
+ write := func() {
+ db.mu.Lock()
+ b := db.batch
+ db.batch = newBatch()
+ db.mu.Unlock()
+ b.Err = db.shed.WriteBatch(b.Batch)
+ close(b.Done)
+ }
+ for {
+ select {
+ case <-db.writeTrigger:
+ write()
+ case <-db.close:
+ // check it there is a batch
+ // left to be written
+ write()
+ return
+ }
+ }
+}
+
+// po computes the proximity order between the address
+// and database base key.
+func (db *DB) po(addr storage.Address) (bin uint8) {
+ return uint8(storage.Proximity(db.baseKey, addr))
+}
+
+// now is a helper function that returns a current unix timestamp
+// in UTC timezone.
+func now() (t int64) {
+ return time.Now().UTC().UnixNano()
+}
diff --git a/swarm/storage/localstore/localstore_test.go b/swarm/storage/localstore/localstore_test.go
new file mode 100644
index 0000000000..1a12c27703
--- /dev/null
+++ b/swarm/storage/localstore/localstore_test.go
@@ -0,0 +1,61 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package localstore
+
+import (
+ "crypto/rand"
+ "io/ioutil"
+ "os"
+ "testing"
+
+ ch "github.com/ethereum/go-ethereum/swarm/chunk"
+ "github.com/ethereum/go-ethereum/swarm/storage"
+)
+
+// newTestDB is a helper function that constructs a
+// temporary database and returns a cleanup function that must
+// be called to remove the data.
+func newTestDB(t *testing.T) (db *DB, cleanupFunc func()) {
+ t.Helper()
+
+ dir, err := ioutil.TempDir("", "shed-test")
+ if err != nil {
+ t.Fatal(err)
+ }
+ cleanupFunc = func() { os.RemoveAll(dir) }
+ baseKey := make([]byte, 32)
+ if _, err := rand.Read(baseKey); err != nil {
+ t.Fatal(err)
+ }
+ db, err = New(dir, baseKey)
+ if err != nil {
+ cleanupFunc()
+ t.Fatal(err)
+ }
+ cleanupFunc = func() {
+ err := db.Close()
+ if err != nil {
+ t.Error(err)
+ }
+ os.RemoveAll(dir)
+ }
+ return db, cleanupFunc
+}
+
+func generateRandomChunk() storage.Chunk {
+ return storage.GenerateRandomChunk(ch.DefaultSize)
+}
diff --git a/swarm/storage/localstore/mode.go b/swarm/storage/localstore/mode.go
new file mode 100644
index 0000000000..63dac3626a
--- /dev/null
+++ b/swarm/storage/localstore/mode.go
@@ -0,0 +1,182 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package localstore
+
+import (
+ "context"
+
+ "github.com/ethereum/go-ethereum/swarm/shed"
+ "github.com/syndtr/goleveldb/leveldb"
+)
+
+// Mode enumerates different modes of access and update
+// operations on a database.
+type Mode int
+
+// Modes of access and update.
+const (
+ ModeSyncing Mode = iota
+ ModeUpload
+ ModeRequest
+ ModeSynced
+ ModeAccess
+ // Q: this mode is not needed,
+ // as it will be used only internally for GC.
+ ModeRemoval
+)
+
+// ModeName returns a descriptive name of a Mode.
+// If the Mode is not know, a blank string is returned.
+func ModeName(m Mode) (name string) {
+ switch m {
+ case ModeSyncing:
+ return "syncing"
+ case ModeUpload:
+ return "upload"
+ case ModeRequest:
+ return "request"
+ case ModeSynced:
+ return "synced"
+ case ModeAccess:
+ return "access"
+ case ModeRemoval:
+ return "removal"
+ }
+ return ""
+}
+
+// access is called by an Accessor with a specific Mode.
+// This function utilizes differnet indexes depending on
+// the Mode.
+func (db *DB) access(mode Mode, item shed.IndexItem) (out shed.IndexItem, err error) {
+ out, err = db.retrievalIndex.Get(item)
+ if err != nil {
+ return out, err
+ }
+ switch mode {
+ case ModeRequest:
+ // update the access counter
+ // Q: can we do this asynchronously
+ return out, db.update(context.TODO(), mode, item)
+ default:
+ // all other modes are not updating the index
+ }
+ return out, nil
+}
+
+// update is called by an Accessor with a specific Mode.
+// This function calles updateBatch to perform operations
+// on indexes and fields within a single batch.
+func (db *DB) update(ctx context.Context, mode Mode, item shed.IndexItem) error {
+ db.mu.RLock()
+ b := db.batch
+ db.mu.RUnlock()
+
+ // check if the database is not closed
+ select {
+ case <-db.close:
+ return ErrDBClosed
+ default:
+ }
+
+ // call the update with the provided mode
+ err := db.updateBatch(b, mode, item)
+ if err != nil {
+ return err
+ }
+ // trigger the writeBatches loop
+ select {
+ case db.writeTrigger <- struct{}{}:
+ default:
+ }
+ // wait for batch to be written and return batch error
+ // this is in order for Put calls to be synchronous
+ select {
+ case <-b.Done:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+ return b.Err
+}
+
+// batch wraps leveldb.Batch extending it with a done channel.
+type batch struct {
+ *leveldb.Batch
+ Done chan struct{} // to signal when batch is written
+ Err error // error resulting from write
+}
+
+// newBatch constructs a new batch.
+func newBatch() *batch {
+ return &batch{
+ Batch: new(leveldb.Batch),
+ Done: make(chan struct{}),
+ }
+}
+
+// updateBatch performs different operations on fields and indexes
+// depending on the provided Mode.
+func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error) {
+ switch mode {
+ case ModeSyncing:
+ // put to indexes: retrieve, pull
+ item.StoreTimestamp = now()
+ item.AccessTimestamp = now()
+ db.retrievalIndex.PutInBatch(b.Batch, item)
+ db.pullIndex.PutInBatch(b.Batch, item)
+ db.sizeCounter.IncInBatch(b.Batch)
+
+ case ModeUpload:
+ // put to indexes: retrieve, push, pull
+ item.StoreTimestamp = now()
+ item.AccessTimestamp = now()
+ db.retrievalIndex.PutInBatch(b.Batch, item)
+ db.pullIndex.PutInBatch(b.Batch, item)
+ db.pushIndex.PutInBatch(b.Batch, item)
+
+ case ModeRequest:
+ // put to indexes: retrieve, gc
+ item.StoreTimestamp = now()
+ item.AccessTimestamp = now()
+ db.retrievalIndex.PutInBatch(b.Batch, item)
+ db.gcIndex.PutInBatch(b.Batch, item)
+
+ case ModeSynced:
+ // delete from push, insert to gc
+ item.StoreTimestamp = now()
+ db.retrievalIndex.PutInBatch(b.Batch, item)
+ db.pushIndex.DeleteInBatch(b.Batch, item)
+ db.gcIndex.PutInBatch(b.Batch, item)
+
+ case ModeAccess:
+ // update accessTimeStamp in retrieve, gc
+ db.gcIndex.DeleteInBatch(b.Batch, item)
+ item.AccessTimestamp = now()
+ db.retrievalIndex.PutInBatch(b.Batch, item)
+ db.gcIndex.PutInBatch(b.Batch, item)
+
+ case ModeRemoval:
+ // delete from retrieve, pull, gc
+ db.retrievalIndex.DeleteInBatch(b.Batch, item)
+ db.pullIndex.DeleteInBatch(b.Batch, item)
+ db.gcIndex.DeleteInBatch(b.Batch, item)
+
+ default:
+ return ErrInvalidMode
+ }
+ return nil
+}