verify_test.go
Rama main · réplica de lectura del repositorio original.
package predastore
import (
"context"
"crypto/sha256"
"errors"
"slices"
"strings"
"sync"
"testing"
"time"
"github.com/mulgadc/predastore/internal/blob"
"github.com/mulgadc/predastore/internal/config"
"github.com/mulgadc/predastore/internal/gate/handlers"
"github.com/mulgadc/predastore/internal/gate/model"
"github.com/mulgadc/predastore/internal/meta"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
type verifyFakeMeta struct {
mu sync.Mutex
rows map[string][]byte
scanErr error
onRecheck func()
rechecked bool
}
func (f *verifyFakeMeta) LeaderScanFrom(_ context.Context, prefix, after string, limit int) ([]meta.Item, error) {
f.mu.Lock()
defer f.mu.Unlock()
if f.scanErr != nil {
return nil, f.scanErr
}
keys := make([]string, 0, len(f.rows))
for k := range f.rows {
if strings.HasPrefix(k, prefix) && k > after {
keys = append(keys, k)
}
}
slices.Sort(keys)
var items []meta.Item
for _, k := range keys {
if len(items) == limit {
break
}
items = append(items, meta.Item{Key: k, Value: f.rows[k]})
}
return items, nil
}
func (f *verifyFakeMeta) LeaderGet(_ context.Context, key string) ([]byte, error) {
f.mu.Lock()
hook := f.onRecheck
if f.rechecked {
hook = nil
}
f.rechecked = true
f.mu.Unlock()
if hook != nil {
hook()
}
f.mu.Lock()
defer f.mu.Unlock()
v, ok := f.rows[key]
if !ok {
return nil, meta.ErrNotFound
}
return v, nil
}
type verifyPos struct {
node NodeID
hash [32]byte
index uint32
}
type verifyHeld struct {
epoch uint64
err error
}
type verifyFakeBlob struct {
mu sync.Mutex
held map[verifyPos]verifyHeld
}
func (f *verifyFakeBlob) set(node NodeID, hash [32]byte, index int, h verifyHeld) {
f.mu.Lock()
defer f.mu.Unlock()
f.held[verifyPos{node, hash, uint32(index)}] = h
}
func (f *verifyFakeBlob) Stat(_ context.Context, node NodeID, req blob.StatRequest) (*blob.StatResponse, error) {
f.mu.Lock()
defer f.mu.Unlock()
h, ok := f.held[verifyPos{node, req.Key, req.Index}]
switch {
case !ok:
return nil, blob.ErrNotFound
case h.err != nil:
return nil, h.err
}
return &blob.StatResponse{Epoch: h.epoch}, nil
}
type verifyFixture struct {
t *testing.T
meta *verifyFakeMeta
blob *verifyFakeBlob
}
func newVerifyFixture(t *testing.T) *verifyFixture {
return &verifyFixture{
t: t,
meta: &verifyFakeMeta{rows: map[string][]byte{}},
blob: &verifyFakeBlob{held: map[verifyPos]verifyHeld{}},
}
}
// object records a 2+1 placement on nodes and, unless told otherwise, has
// every node hold its shard at the record's epoch.
func (x *verifyFixture) object(name string, epoch uint64, nodes ...NodeID) ([32]byte, handlers.ObjectToShardNodes) {
x.t.Helper()
hash := sha256.Sum256([]byte(name))
place := handlers.ObjectToShardNodes{
DataShardNodes: nodes[:2], ParityShardNodes: nodes[2:], Size: 4096, WriteEpoch: epoch,
}
x.record(hash, place)
for i, n := range nodes {
x.blob.set(n, hash, i, verifyHeld{epoch: epoch})
}
return hash, place
}
func (x *verifyFixture) record(hash [32]byte, place handlers.ObjectToShardNodes) {
x.t.Helper()
raw, err := handlers.EncodePlacement(place)
require.NoError(x.t, err)
x.meta.mu.Lock()
defer x.meta.mu.Unlock()
x.meta.rows[handlers.TableKey(model.TableObjects, string(hash[:]))] = raw
}
// run verifies with a recheck short enough for a test to wait out.
func (x *verifyFixture) run(opts VerifyOptions) (ShardReport, error) {
opts.Recheck = time.Millisecond
return verifyShards(x.t.Context(), x.meta, x.blob, []NodeID{1, 2, 3, 4}, opts)
}
func TestVerifyShardsCountsEachPositionAgainstTheRecord(t *testing.T) {
t.Parallel()
x := newVerifyFixture(t)
x.object("full", 10, 1, 2, 3)
damaged, _ := x.object("damaged", 20, 2, 3, 4)
x.blob.set(3, damaged, 1, verifyHeld{epoch: 19})
delete(x.blob.held, verifyPos{4, damaged, 2})
x.meta.rows[handlers.TableKey(model.TableObjects, "arn:aws:s3:::bucket/damaged")] = damaged[:]
unreadable, _ := x.object("unreadable", 30, 1, 3, 4)
x.blob.set(4, unreadable, 2, verifyHeld{err: errors.New("stat: disk on fire")})
x.record(sha256.Sum256([]byte("empty")), handlers.ObjectToShardNodes{
DataShardNodes: []config.NodeID{1, 2}, ParityShardNodes: []config.NodeID{3}, WriteEpoch: 40,
})
undecodable := sha256.Sum256([]byte("undecodable"))
x.meta.rows[handlers.TableKey(model.TableObjects, string(undecodable[:]))] = []byte("junk")
report, err := x.run(VerifyOptions{PageSize: 2})
require.NoError(t, err)
assert.Equal(t, 3, report.Records)
assert.Equal(t, 1, report.Empty)
assert.Equal(t, 1, report.Undecodable)
assert.Equal(t, 3, report.Checked)
assert.Equal(t, 1, report.Full)
assert.Equal(t, map[int]int{3: 1, 2: 1, 1: 1}, report.ByMatch)
require.Len(t, report.NonFull, 2)
worst := report.NonFull[0]
assert.Equal(t, 1, worst.Match, "the least intact object is listed first")
assert.Equal(t, "bucket/damaged", worst.Name)
assert.Equal(t, []string{ShardMatch, ShardOlder, ShardMissing},
[]string{worst.Shards[0].State, worst.Shards[1].State, worst.Shards[2].State})
assert.Equal(t, uint64(19), worst.Shards[1].HeldEpoch)
assert.Equal(t, ShardError, report.NonFull[1].Shards[2].State)
assert.Contains(t, report.NonFull[1].Shards[2].Error, "disk on fire")
require.Len(t, report.Nodes, 4)
assert.Equal(t, NodeShardCounts{Node: 3, Owned: 3, Match: 2, Older: 1}, report.Nodes[2])
assert.Equal(t, NodeShardCounts{Node: 4, Owned: 2, Missing: 1, Error: 1}, report.Nodes[3])
}
// TestVerifyShardsRechecksAnObjectWrittenMidScan keeps a write in flight, or
// a delete, from being reported as damage.
func TestVerifyShardsRechecksAnObjectWrittenMidScan(t *testing.T) {
t.Parallel()
x := newVerifyFixture(t)
rewritten, place := x.object("rewritten", 10, 1, 2, 3)
x.blob.set(3, rewritten, 2, verifyHeld{epoch: 11})
deleted, _ := x.object("deleted", 10, 2, 3, 4)
delete(x.blob.held, verifyPos{2, deleted, 0})
x.meta.onRecheck = func() {
place.WriteEpoch = 11
x.record(rewritten, place)
for i, n := range place.AllNodes() {
x.blob.set(n, rewritten, i, verifyHeld{epoch: 11})
}
x.meta.mu.Lock()
delete(x.meta.rows, handlers.TableKey(model.TableObjects, string(deleted[:])))
x.meta.mu.Unlock()
}
report, err := x.run(VerifyOptions{})
require.NoError(t, err)
assert.Equal(t, 2, report.Records)
assert.Equal(t, 1, report.Vanished)
assert.Equal(t, 1, report.Checked)
assert.Equal(t, 1, report.Full)
assert.Empty(t, report.NonFull)
}
func TestVerifyShardsFailsWhenTheScanFails(t *testing.T) {
t.Parallel()
x := newVerifyFixture(t)
x.meta.scanErr = meta.ErrNoLeaderRead
_, err := x.run(VerifyOptions{})
require.ErrorIs(t, err, meta.ErrNoLeaderRead)
}
func TestVerifyShardsNeedsMetaAndBlobNodes(t *testing.T) {
t.Parallel()
cfg := &Config{Hosts: []config.Host{{ID: 1, Addr: "127.0.0.1", Nodes: []config.Node{{ID: 1, Role: config.RoleBlob, Port: 1}}}}}
_, err := VerifyShards(t.Context(), cfg, nil, VerifyOptions{})
require.ErrorContains(t, err, "0 meta and 1 blob nodes")
}