func benchmarkPutUpload(b *testing.B, o *Options, count, maxParallelUploads int) {
	b.StopTimer()
	db := newTestDB(b, o)
	chunks := make([]swarm.Chunk, count)
	for i := 0; i < count; i++ {
		chunks[i] = generateTestRandomChunk()
	}
	errs := make(chan error)
	b.StartTimer()
	go func() {
		sem := make(chan struct{}, maxParallelUploads)
		for i := 0; i < count; i++ {
			sem <- struct{}{}
			go func(i int) {
				defer func() { <-sem }()
				_, err := db.Put(context.Background(), storage.ModePutUpload, chunks[i])
				errs <- err
			}(i)
		}
	}()
	for i := 0; i < count; i++ {
		err := <-errs
		if err != nil {
			b.Fatal(err)
		}
	}
}
benchmarkPutUpload runs a benchmark by uploading a specific number
of chunks with specified max parallel uploads.
package localstore
import (
	"bytes"
	"context"
	"encoding/binary"
	"errors"
	"fmt"
	"sync"
	"testing"
	"time"
	"github.com/ethsana/sana/pkg/postage"
	postagetesting "github.com/ethsana/sana/pkg/postage/testing"
	"github.com/ethsana/sana/pkg/shed"
	"github.com/ethsana/sana/pkg/storage"
	"github.com/ethsana/sana/pkg/swarm"
	"github.com/syndtr/goleveldb/leveldb"
)
var putModes = []storage.ModePut{
	storage.ModePutRequest,
	storage.ModePutRequestPin,
	storage.ModePutSync,
	storage.ModePutUpload,
	storage.ModePutUploadPin,
	storage.ModePutRequestCache,
}
func TestModePutRequest(t *testing.T) {
	t.Cleanup(setWithinRadiusFunc(func(_ *DB, _ shed.Item) bool { return false }))
	for _, tc := range multiChunkTestCases {
		t.Run(tc.name, func(t *testing.T) {
			db := newTestDB(t, nil)
			chunks := generateTestRandomChunks(tc.count)
			unreserveChunkBatch(t, db, 0, chunks...)
			var storeTimestamp int64
			t.Run("first put", func(t *testing.T) {
				wantTimestamp := time.Now().UTC().UnixNano()
				defer setNow(func() (t int64) {
					return wantTimestamp
				})()
				storeTimestamp = wantTimestamp
				_, err := db.Put(context.Background(), storage.ModePutRequest, chunks...)
				if err != nil {
					t.Fatal(err)
				}
				for _, ch := range chunks {
					newRetrieveIndexesTestWithAccess(db, ch, wantTimestamp, wantTimestamp)(t)
				}
				newItemsCountTest(db.gcIndex, tc.count)(t)
				newItemsCountTest(db.pullIndex, tc.count)(t)
				newItemsCountTest(db.postageIndexIndex, tc.count)(t)
				newIndexGCSizeTest(db)(t)
			})
			t.Run("second put", func(t *testing.T) {
				wantTimestamp := time.Now().UTC().UnixNano()
				defer setNow(func() (t int64) {
					return wantTimestamp
				})()
				_, err := db.Put(context.Background(), storage.ModePutRequest, chunks...)
				if err != nil {
					t.Fatal(err)
				}
				for _, ch := range chunks {
					newRetrieveIndexesTestWithAccess(db, ch, storeTimestamp, storeTimestamp)(t)
				}
				newItemsCountTest(db.gcIndex, tc.count)(t)
				newItemsCountTest(db.pullIndex, tc.count)(t)
				newItemsCountTest(db.postageIndexIndex, tc.count)(t)
				newIndexGCSizeTest(db)(t)
			})
		})
	}
}
func TestModePutRequestPin(t *testing.T) {
	t.Cleanup(setWithinRadiusFunc(func(_ *DB, _ shed.Item) bool { return false }))
	for _, tc := range multiChunkTestCases {
		t.Run(tc.name, func(t *testing.T) {
			db := newTestDB(t, nil)
			chunks := generateTestRandomChunks(tc.count)
			unreserveChunkBatch(t, db, 0, chunks...)
			wantTimestamp := time.Now().UTC().UnixNano()
			defer setNow(func() (t int64) {
				return wantTimestamp
			})()
			_, err := db.Put(context.Background(), storage.ModePutRequestPin, chunks...)
			if err != nil {
				t.Fatal(err)
			}
			for _, ch := range chunks {
				newRetrieveIndexesTestWithAccess(db, ch, wantTimestamp, wantTimestamp)(t)
				newPinIndexTest(db, ch, nil)(t)
			}
			newItemsCountTest(db.postageChunksIndex, tc.count)(t)
			newItemsCountTest(db.postageIndexIndex, tc.count)(t)
			newItemsCountTest(db.gcIndex, 0)(t)
			newIndexGCSizeTest(db)(t)
		})
	}
}
func TestModePutRequestCache(t *testing.T) {
	t.Cleanup(setWithinRadiusFunc(func(_ *DB, _ shed.Item) bool { return true }))
	for _, tc := range multiChunkTestCases {
		t.Run(tc.name, func(t *testing.T) {
			db := newTestDB(t, nil)
			var chunks []swarm.Chunk
			for i := 0; i < tc.count; i++ {
				chunk := generateTestRandomChunkAt(swarm.NewAddress(db.baseKey), 2)
				chunks = append(chunks, chunk)
			}
			unreserveChunkBatch(t, db, 2, chunks...)
			wantTimestamp := time.Now().UTC().UnixNano()
			defer setNow(func() (t int64) {
				return wantTimestamp
			})()
			_, err := db.Put(context.Background(), storage.ModePutRequestCache, chunks...)
			if err != nil {
				t.Fatal(err)
			}
			for _, ch := range chunks {
				newRetrieveIndexesTestWithAccess(db, ch, wantTimestamp, wantTimestamp)(t)
				newPinIndexTest(db, ch, leveldb.ErrNotFound)(t)
			}
			newItemsCountTest(db.postageChunksIndex, tc.count)(t)
			newItemsCountTest(db.postageIndexIndex, tc.count)(t)
			newItemsCountTest(db.gcIndex, tc.count)(t)
			newIndexGCSizeTest(db)(t)
		})
	}
}
func TestModePutSync(t *testing.T) {
	t.Cleanup(setWithinRadiusFunc(func(_ *DB, _ shed.Item) bool { return false }))
	for _, tc := range multiChunkTestCases {
		t.Run(tc.name, func(t *testing.T) {
			db := newTestDB(t, nil)
			wantTimestamp := time.Now().UTC().UnixNano()
			defer setNow(func() (t int64) {
				return wantTimestamp
			})()
			chunks := generateTestRandomChunks(tc.count)
			unreserveChunkBatch(t, db, 0, chunks...)
			_, err := db.Put(context.Background(), storage.ModePutSync, chunks...)
			if err != nil {
				t.Fatal(err)
			}
			binIDs := make(map[uint8]uint64)
			for _, ch := range chunks {
				po := db.po(ch.Address())
				binIDs[po]++
				newRetrieveIndexesTestWithAccess(db, ch, wantTimestamp, wantTimestamp)(t)
				newPullIndexTest(db, ch, binIDs[po], nil)(t)
				newPinIndexTest(db, ch, leveldb.ErrNotFound)(t)
				newIndexGCSizeTest(db)(t)
			}
			newItemsCountTest(db.postageChunksIndex, tc.count)(t)
			newItemsCountTest(db.postageIndexIndex, tc.count)(t)
			newItemsCountTest(db.gcIndex, tc.count)(t)
			newIndexGCSizeTest(db)(t)
		})
	}
}
func TestModePutUpload(t *testing.T) {
	for _, tc := range multiChunkTestCases {
		t.Run(tc.name, func(t *testing.T) {
			db := newTestDB(t, nil)
			wantTimestamp := time.Now().UTC().UnixNano()
			defer setNow(func() (t int64) {
				return wantTimestamp
			})()
			chunks := generateTestRandomChunks(tc.count)
			unreserveChunkBatch(t, db, 0, chunks...)
			_, err := db.Put(context.Background(), storage.ModePutUpload, chunks...)
			if err != nil {
				t.Fatal(err)
			}
			binIDs := make(map[uint8]uint64)
			for _, ch := range chunks {
				po := db.po(ch.Address())
				binIDs[po]++
				newRetrieveIndexesTest(db, ch, wantTimestamp, 0)(t)
				newPullIndexTest(db, ch, binIDs[po], nil)(t)
				newPushIndexTest(db, ch, wantTimestamp, nil)(t)
				newPinIndexTest(db, ch, leveldb.ErrNotFound)(t)
			}
			newItemsCountTest(db.postageIndexIndex, tc.count)(t)
			newIndexGCSizeTest(db)(t)
		})
	}
}
func TestModePutUploadPin(t *testing.T) {
	for _, tc := range multiChunkTestCases {
		t.Run(tc.name, func(t *testing.T) {
			db := newTestDB(t, nil)
			wantTimestamp := time.Now().UTC().UnixNano()
			defer setNow(func() (t int64) {
				return wantTimestamp
			})()
			chunks := generateTestRandomChunks(tc.count)
			unreserveChunkBatch(t, db, 0, chunks...)
			_, err := db.Put(context.Background(), storage.ModePutUploadPin, chunks...)
			if err != nil {
				t.Fatal(err)
			}
			binIDs := make(map[uint8]uint64)
			for _, ch := range chunks {
				po := db.po(ch.Address())
				binIDs[po]++
				newRetrieveIndexesTest(db, ch, wantTimestamp, 0)(t)
				newPullIndexTest(db, ch, binIDs[po], nil)(t)
				newPushIndexTest(db, ch, wantTimestamp, nil)(t)
				newPinIndexTest(db, ch, nil)(t)
			}
			newItemsCountTest(db.postageIndexIndex, tc.count)(t)
			newIndexGCSizeTest(db)(t)
		})
	}
}
func TestModePutUpload_parallel(t *testing.T) {
	for _, tc := range []struct {
		name  string
		count int
	}{
		{
			name:  "one",
			count: 1,
		},
		{
			name:  "two",
			count: 2,
		},
		{
			name:  "eight",
			count: 8,
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			db := newTestDB(t, nil)
			uploadsCount := 100
			workerCount := 100
			chunksChan := make(chan []swarm.Chunk)
			errChan := make(chan error)
			doneChan := make(chan struct{})
			defer close(doneChan)
			for i := 0; i < workerCount; i++ {
				go func(i int) {
					for {
						select {
						case chunks, ok := <-chunksChan:
							if !ok {
								return
							}
							_, err := db.Put(context.Background(), storage.ModePutUpload, chunks...)
							select {
							case errChan <- err:
							case <-doneChan:
							}
						case <-doneChan:
							return
						}
					}
				}(i)
			}
			chunks := make([]swarm.Chunk, 0)
			var chunksMu sync.Mutex
			go func() {
				for i := 0; i < uploadsCount; i++ {
					chs := generateTestRandomChunks(tc.count)
					unreserveChunkBatch(t, db, 0, chunks...)
					select {
					case chunksChan <- chs:
					case <-doneChan:
						return
					}
					chunksMu.Lock()
					chunks = append(chunks, chs...)
					chunksMu.Unlock()
				}
				close(chunksChan)
			}()
			for i := 0; i < uploadsCount; i++ {
				err := <-errChan
				if err != nil {
					t.Fatal(err)
				}
			}
			chunksMu.Lock()
			defer chunksMu.Unlock()
			for _, ch := range chunks {
				got, err := db.Get(context.Background(), storage.ModeGetRequest, ch.Address())
				if err != nil {
					t.Fatal(err)
				}
				if !bytes.Equal(got.Data(), ch.Data()) {
					t.Fatalf("got chunk %s data %x, want %x", ch.Address(), got.Data(), ch.Data())
				}
			}
		})
	}
}
func TestModePut_sameChunk(t *testing.T) {
	for _, tc := range multiChunkTestCases {
		t.Run(tc.name, func(t *testing.T) {
			chunks := generateTestRandomChunks(tc.count)
			for _, tcn := range []struct {
				name      string
				mode      storage.ModePut
				pullIndex bool
				pushIndex bool
			}{
				{
					name:      "ModePutRequest",
					mode:      storage.ModePutRequest,
					pullIndex: true,
					pushIndex: false,
				},
				{
					name:      "ModePutRequestPin",
					mode:      storage.ModePutRequest,
					pullIndex: true,
					pushIndex: false,
				},
				{
					name:      "ModePutRequestCache",
					mode:      storage.ModePutRequestCache,
					pullIndex: false,
					pushIndex: false,
				},
				{
					name:      "ModePutUpload",
					mode:      storage.ModePutUpload,
					pullIndex: true,
					pushIndex: true,
				},
				{
					name:      "ModePutSync",
					mode:      storage.ModePutSync,
					pullIndex: true,
					pushIndex: false,
				},
			} {
				t.Run(tcn.name, func(t *testing.T) {
					db := newTestDB(t, nil)
					unreserveChunkBatch(t, db, 0, chunks...)
					for i := 0; i < 10; i++ {
						exist, err := db.Put(context.Background(), tcn.mode, chunks...)
						if err != nil {
							t.Fatal(err)
						}
						for _, exists := range exist {
							switch exists {
							case false:
								if i != 0 {
									t.Fatal("should not exist only on first Put")
								}
							case true:
								if i == 0 {
									t.Fatal("should exist on all cases other than the first one")
								}
							}
						}
						count := func(b bool) (c int) {
							if b {
								return tc.count
							}
							return 0
						}
						newItemsCountTest(db.retrievalDataIndex, tc.count)(t)
						newItemsCountTest(db.pullIndex, count(tcn.pullIndex))(t)
						newItemsCountTest(db.pushIndex, count(tcn.pushIndex))(t)
						newIndexGCSizeTest(db)(t)
					}
				})
			}
		})
	}
}
func TestModePut_SameStamp(t *testing.T) {
	ctx := context.Background()
	stamp := postagetesting.MustNewStamp()
	ts := time.Now().Unix()
	for _, modeTc1 := range putModes {
		for _, modeTc2 := range putModes {
			for _, tc := range []struct {
				persistChunk swarm.Chunk
				discardChunk swarm.Chunk
			}{
				{
					persistChunk: generateChunkWithTimestamp(stamp, ts),
					discardChunk: generateChunkWithTimestamp(stamp, ts),
				},
				{
					persistChunk: generateChunkWithTimestamp(stamp, ts+1),
					discardChunk: generateChunkWithTimestamp(stamp, ts),
				},
				{
					persistChunk: generateChunkWithTimestamp(stamp, ts),
					discardChunk: generateChunkWithTimestamp(stamp, ts-1),
				},
			} {
				t.Run(modeTc1.String()+modeTc2.String(), func(t *testing.T) {
					db := newTestDB(t, nil)
					unreserveChunkBatch(t, db, 0, tc.persistChunk, tc.discardChunk)
					_, err := db.Put(ctx, modeTc1, tc.persistChunk)
					if err != nil {
						t.Fatal(err)
					}
					_, err = db.Put(ctx, modeTc2, tc.discardChunk)
					if err != nil {
						t.Fatal(err)
					}
					newItemsCountTest(db.retrievalDataIndex, 1)(t)
					newItemsCountTest(db.postageChunksIndex, 1)(t)
					newItemsCountTest(db.postageRadiusIndex, 1)(t)
					newItemsCountTest(db.postageIndexIndex, 1)(t)
					if modeTc1 != storage.ModePutRequestCache {
						newItemsCountTest(db.pullIndex, 1)(t)
					}
					_, err = db.Get(ctx, storage.ModeGetLookup, tc.persistChunk.Address())
					if err != nil {
						t.Fatalf("expected no error, got %v", err)
					}
					_, err = db.Get(ctx, storage.ModeGetLookup, tc.discardChunk.Address())
					if !errors.Is(err, storage.ErrNotFound) {
						t.Fatalf("expected %v, got %v", storage.ErrNotFound, err)
					}
				})
			}
		}
	}
}
func TestModePut_ImmutableStamp(t *testing.T) {
	ctx := context.Background()
	stamp := postagetesting.MustNewStamp()
	ts := time.Now().Unix()
	for _, modeTc1 := range putModes {
		for _, modeTc2 := range putModes {
			for _, tc := range []struct {
				name         string
				persistChunk swarm.Chunk
				discardChunk swarm.Chunk
			}{
				{
					name:         "same timestamps",
					persistChunk: generateImmutableChunkWithTimestamp(stamp, ts),
					discardChunk: generateImmutableChunkWithTimestamp(stamp, ts),
				},
				{
					name:         "higher timestamp",
					persistChunk: generateImmutableChunkWithTimestamp(stamp, ts),
					discardChunk: generateImmutableChunkWithTimestamp(stamp, ts+1),
				},
				{
					name:         "higher timestamp first",
					persistChunk: generateImmutableChunkWithTimestamp(stamp, ts+1),
					discardChunk: generateImmutableChunkWithTimestamp(stamp, ts),
				},
			} {
				testName := fmt.Sprintf("%s %s %s", modeTc1.String(), modeTc2.String(), tc.name)
				t.Run(testName, func(t *testing.T) {
					db := newTestDB(t, nil)
					unreserveChunkBatch(t, db, 0, tc.persistChunk, tc.discardChunk)
					_, err := db.Put(ctx, modeTc1, tc.persistChunk)
					if err != nil {
						t.Fatal(err)
					}
					_, err = db.Put(ctx, modeTc2, tc.discardChunk)
					if !errors.Is(err, ErrOverwrite) {
						t.Fatalf("expected overwrite error on immutable stamp got %v", err)
					}
					newItemsCountTest(db.retrievalDataIndex, 1)(t)
					newItemsCountTest(db.postageChunksIndex, 1)(t)
					newItemsCountTest(db.postageRadiusIndex, 1)(t)
					newItemsCountTest(db.postageIndexIndex, 1)(t)
					if modeTc1 != storage.ModePutRequestCache {
						newItemsCountTest(db.pullIndex, 1)(t)
					}
					_, err = db.Get(ctx, storage.ModeGetLookup, tc.persistChunk.Address())
					if err != nil {
						t.Fatalf("expected no error, got %v", err)
					}
					_, err = db.Get(ctx, storage.ModeGetLookup, tc.discardChunk.Address())
					if !errors.Is(err, storage.ErrNotFound) {
						t.Fatalf("expected %v, got %v", storage.ErrNotFound, err)
					}
				})
			}
		}
	}
}
func generateChunkWithTimestamp(stamp *postage.Stamp, timestamp int64) swarm.Chunk {
	tsBuf := make([]byte, 8)
	binary.BigEndian.PutUint64(tsBuf, uint64(timestamp))
	chunk := generateTestRandomChunk()
	return chunk.WithStamp(postage.NewStamp(stamp.BatchID(), stamp.Index(), tsBuf, stamp.Sig()))
}
func generateImmutableChunkWithTimestamp(stamp *postage.Stamp, timestamp int64) swarm.Chunk {
	return generateChunkWithTimestamp(stamp, timestamp).WithBatch(4, 12, 8, true)
}
func TestPutDuplicateChunks(t *testing.T) {
	for _, mode := range []storage.ModePut{
		storage.ModePutUpload,
		storage.ModePutRequest,
		storage.ModePutSync,
	} {
		t.Run(mode.String(), func(t *testing.T) {
			db := newTestDB(t, nil)
			ch := generateTestRandomChunk()
			unreserveChunkBatch(t, db, 0, ch)
			exist, err := db.Put(context.Background(), mode, ch, ch)
			if err != nil {
				t.Fatal(err)
			}
			if exist[0] {
				t.Error("first chunk should not exist")
			}
			if !exist[1] {
				t.Error("second chunk should exist")
			}
			newItemsCountTest(db.retrievalDataIndex, 1)(t)
			got, err := db.Get(context.Background(), storage.ModeGetLookup, ch.Address())
			if err != nil {
				t.Fatal(err)
			}
			if !got.Address().Equal(ch.Address()) {
				t.Errorf("got chunk address %s, want %s", got.Address(), ch.Address())
			}
		})
	}
}
func BenchmarkPutUpload(b *testing.B) {
	for _, count := range []int{
		100,
		1000,
		10000,
		100000,
	} {
		for _, maxParallelUploads := range []int{
			1,
			2,
			4,
			8,
			16,
			32,
		} {
			name := fmt.Sprintf("count %v parallel %v", count, maxParallelUploads)
			b.Run(name, func(b *testing.B) {
				for n := 0; n < b.N; n++ {
					benchmarkPutUpload(b, nil, count, maxParallelUploads)
				}
			})
		}
	}
}