diff --git a/swarm/storage/localstore/accessor_test.go b/swarm/storage/localstore/accessor_test.go index f09eb82ee1..439f7d9db3 100644 --- a/swarm/storage/localstore/accessor_test.go +++ b/swarm/storage/localstore/accessor_test.go @@ -52,7 +52,6 @@ func testAccessors(t *testing.T, db *DB) { ModeSyncing, ModeUpload, ModeRequest, - modeAccess, } { t.Run(ModeName(m), func(t *testing.T) { a := db.Accessor(m) @@ -93,6 +92,30 @@ func testAccessors(t *testing.T, db *DB) { } }) + // Access mode is a special as it does not store the chunk + // in the database. + t.Run(ModeName(modeAccess), 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) + } + + a = db.Accessor(modeAccess) + + 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) { diff --git a/swarm/storage/localstore/mode.go b/swarm/storage/localstore/mode.go index d914a8c4ed..dbebe91333 100644 --- a/swarm/storage/localstore/mode.go +++ b/swarm/storage/localstore/mode.go @@ -70,16 +70,19 @@ func (db *DB) access(mode Mode, item shed.IndexItem) (out shed.IndexItem, err er return out, err } } else { + // No need to get access timestamp here as it is used + // only for some of Modes in update and access time + // is not property of the chunk returned by the Accessor.Get. out, err = db.retrievalDataIndex.Get(item) if err != nil { return out, err } } switch mode { - case ModeRequest: + case ModeRequest, modeAccess: // update the access counter // Q: can we do this asynchronously - return out, db.update(context.TODO(), mode, item) + return out, db.update(context.TODO(), mode, out) default: // all other modes are not updating the index } @@ -180,6 +183,9 @@ func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error) case ModeSynced: // delete from push, insert to gc item.StoreTimestamp = now() + // need to get access timestamp here as it is not + // provided by the access function, and it is not + // a property of a chunk provided to Accessor.Put. if db.useRetrievalCompositeIndex { i, err := db.retrievalCompositeIndex.Get(item) switch err { @@ -206,18 +212,67 @@ func (db *DB) updateBatch(b *batch, mode Mode, item shed.IndexItem) (err error) case modeAccess: // update accessTimeStamp in retrieve, gc + + // need to get access timestamp here as it is not + // provided by the access function, and it is not + // a property of a chunk provided to Accessor.Put. + if db.useRetrievalCompositeIndex { + i, err := db.retrievalCompositeIndex.Get(item) + if err != nil { + return err + } + item.AccessTimestamp = i.AccessTimestamp + } else { + i, err := db.retrievalAccessIndex.Get(item) + switch err { + case nil: + item.AccessTimestamp = i.AccessTimestamp + case leveldb.ErrNotFound: + item.AccessTimestamp = now() + default: + return err + } + } + // delete current entry from the gc index db.gcIndex.DeleteInBatch(b.Batch, item) + // update access timestamp item.AccessTimestamp = now() + // update retrieve access index if db.useRetrievalCompositeIndex { db.retrievalCompositeIndex.PutInBatch(b.Batch, item) } else { - db.retrievalDataIndex.PutInBatch(b.Batch, item) db.retrievalAccessIndex.PutInBatch(b.Batch, item) } + // add new entry to gc index db.gcIndex.PutInBatch(b.Batch, item) case modeRemoval: // delete from retrieve, pull, gc + + // need to get access timestamp here as it is not + // provided by the access function, and it is not + // a property of a chunk provided to Accessor.Put. + if db.useRetrievalCompositeIndex { + i, err := db.retrievalCompositeIndex.Get(item) + switch err { + case nil: + item.AccessTimestamp = i.AccessTimestamp + case leveldb.ErrNotFound: + item.AccessTimestamp = now() + default: + return err + } + } else { + i, err := db.retrievalAccessIndex.Get(item) + switch err { + case nil: + item.AccessTimestamp = i.AccessTimestamp + case leveldb.ErrNotFound: + item.AccessTimestamp = now() + default: + return err + } + } if db.useRetrievalCompositeIndex { db.retrievalCompositeIndex.DeleteInBatch(b.Batch, item) } else { diff --git a/swarm/storage/localstore/mode_test.go b/swarm/storage/localstore/mode_test.go index 8d72f829a6..50a9db7938 100644 --- a/swarm/storage/localstore/mode_test.go +++ b/swarm/storage/localstore/mode_test.go @@ -47,13 +47,14 @@ func TestModeSyncing_withRetrievalCompositeIndex(t *testing.T) { testModeSyncingValues(t, db) } -// testModeSyncingValues validates ModeSyncing on the provided DB. +// testModeSyncingValues validates ModeSyncing index values on the provided DB. func testModeSyncingValues(t *testing.T, db *DB) { a := db.Accessor(ModeSyncing) chunk := generateRandomChunk() wantTimestamp := time.Now().UTC().UnixNano() + defer func(n func() int64) { now = n }(now) now = func() (t int64) { return wantTimestamp } @@ -96,13 +97,14 @@ func TestModeUpload_withRetrievalCompositeIndex(t *testing.T) { testModeUploadValues(t, db) } -// testModeUploadValues validates ModeUpload on the provided DB. +// testModeUploadValues validates ModeUpload index values on the provided DB. func testModeUploadValues(t *testing.T, db *DB) { a := db.Accessor(ModeUpload) chunk := generateRandomChunk() wantTimestamp := time.Now().UTC().UnixNano() + defer func(n func() int64) { now = n }(now) now = func() (t int64) { return wantTimestamp } @@ -147,13 +149,14 @@ func TestModeRequest_withRetrievalCompositeIndex(t *testing.T) { testModeRequestValues(t, db) } -// testModeRequestValues validates ModeRequest on the provided DB. +// testModeRequestValues validates ModeRequest index values on the provided DB. func testModeRequestValues(t *testing.T, db *DB) { a := db.Accessor(ModeRequest) chunk := generateRandomChunk() wantTimestamp := time.Now().UTC().UnixNano() + defer func(n func() int64) { now = n }(now) now = func() (t int64) { return wantTimestamp } @@ -187,13 +190,14 @@ func TestModeSynced_withRetrievalCompositeIndex(t *testing.T) { testModeSyncedValues(t, db) } -// testModeSyncedValues validates ModeSynced on the provided DB. +// testModeSyncedValues validates ModeSynced index values on the provided DB. func testModeSyncedValues(t *testing.T, db *DB) { a := db.Accessor(ModeSyncing) chunk := generateRandomChunk() wantTimestamp := time.Now().UTC().UnixNano() + defer func(n func() int64) { now = n }(now) now = func() (t int64) { return wantTimestamp } @@ -217,6 +221,92 @@ func testModeSyncedValues(t *testing.T, db *DB) { t.Run("gc index", testGCIndexValues(db, chunk, wantTimestamp, wantTimestamp)) } +// TestModeAccess validates internal data operations and state +// for ModeAccess on DB with default configuration. +func TestModeAccess(t *testing.T) { + db, cleanupFunc := newTestDB(t) + defer cleanupFunc() + + testModeAccessValues(t, db) +} + +// TestModeAccess_withRetrievalCompositeIndex validates internal +// data operations and state for ModeAccess on DB with +// retrieval composite index enabled. +func TestModeAccess_withRetrievalCompositeIndex(t *testing.T) { + db, cleanupFunc := newTestDB(t, WithRetrievalCompositeIndex(true)) + defer cleanupFunc() + + testModeAccessValues(t, db) +} + +// testModeAccessValues validates ModeAccess index values on the provided DB. +func testModeAccessValues(t *testing.T, db *DB) { + a := db.Accessor(ModeUpload) + + chunk := generateRandomChunk() + + uploadTimestamp := time.Now().UTC().UnixNano() + defer func(n func() int64) { now = n }(now) + now = func() (t int64) { + return uploadTimestamp + } + + err := a.Put(context.Background(), chunk) + if err != nil { + t.Fatal(err) + } + + a = db.Accessor(modeAccess) + + t.Run("first get", func(t *testing.T) { + got, err := a.Get(context.Background(), chunk.Address()) + if err != nil { + t.Fatal(err) + } + + if !bytes.Equal(chunk.Address(), got.Address()) { + t.Errorf("got chunk address %x, want %s", chunk.Address(), got.Address()) + } + + if !bytes.Equal(chunk.Data(), got.Data()) { + t.Errorf("got chunk data %x, want %s", chunk.Data(), got.Data()) + } + + t.Run("retrieve indexes", testRetrieveIndexesValuesWithAccess(db, chunk, uploadTimestamp, uploadTimestamp)) + + t.Run("gc index", testGCIndexValues(db, chunk, uploadTimestamp, uploadTimestamp)) + + t.Run("gc index count", testGCIndexCount(db, 1)) + }) + + t.Run("second get", func(t *testing.T) { + accessTimestamp := time.Now().UTC().UnixNano() + now = func() (t int64) { + return accessTimestamp + } + + got, err := a.Get(context.Background(), chunk.Address()) + if err != nil { + t.Fatal(err) + } + + if !bytes.Equal(chunk.Address(), got.Address()) { + t.Errorf("got chunk address %x, want %s", chunk.Address(), got.Address()) + } + + if !bytes.Equal(chunk.Data(), got.Data()) { + t.Errorf("got chunk data %x, want %s", chunk.Data(), got.Data()) + } + + t.Run("retrieve indexes", testRetrieveIndexesValuesWithAccess(db, chunk, uploadTimestamp, accessTimestamp)) + + t.Run("gc index", testGCIndexValues(db, chunk, uploadTimestamp, accessTimestamp)) + + t.Run("gc index count", testGCIndexCount(db, 1)) + }) +} + // testRetrieveIndexesValues returns a test function that validates if the right // chunk values are in the retrieval indexes. func testRetrieveIndexesValues(db *DB, chunk storage.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) { @@ -319,6 +409,21 @@ func testGCIndexValues(db *DB, chunk storage.Chunk, storeTimestamp, accessTimest } } +// testGCIndexCount returns a test function that validates if +// gc index contains expected number of key/value pairs. +func testGCIndexCount(db *DB, want int) func(t *testing.T) { + return func(t *testing.T) { + var c int + db.gcIndex.IterateAll(func(item shed.IndexItem) (stop bool, err error) { + c++ + return + }) + if c != want { + t.Errorf("got %v item in gc index, want %v", c, want) + } + } +} + // testSizeCounter returns a test function that validates the expected // value from sizeCounter field. func testSizeCounter(db *DB, wantSize uint64) func(t *testing.T) {