diff --git a/weed/s3api/s3lifecycle/lifecycletest/fakeserver.go b/weed/s3api/s3lifecycle/lifecycletest/fakeserver.go new file mode 100644 index 000000000..dd55e50a6 --- /dev/null +++ b/weed/s3api/s3lifecycle/lifecycletest/fakeserver.go @@ -0,0 +1,159 @@ +// Package lifecycletest provides reusable test doubles for the lifecycle +// worker pipeline. The pieces here let component-level tests stand up the +// gRPC boundary the worker dials at runtime without pulling in a real +// S3ApiServer or filer. +package lifecycletest + +import ( + "context" + "sync" + + "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" + "google.golang.org/protobuf/proto" +) + +// Outcome is what the fake returns for a single LifecycleDelete call. +// Code is required; Reason is echoed verbatim into LifecycleDeleteResponse. +type Outcome struct { + Code s3_lifecycle_pb.LifecycleDeleteOutcome + Reason string +} + +// Done, NoopResolved, RetryLater, Blocked, SkippedObjectLock are +// constructors for the common outcomes; tests can also build Outcome +// values directly when they need a specific Reason. +func Done() Outcome { + return Outcome{Code: s3_lifecycle_pb.LifecycleDeleteOutcome_DONE} +} +func NoopResolved(reason string) Outcome { + return Outcome{Code: s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED, Reason: reason} +} +func RetryLater(reason string) Outcome { + return Outcome{Code: s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER, Reason: reason} +} +func Blocked(reason string) Outcome { + return Outcome{Code: s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED, Reason: reason} +} +func SkippedObjectLock(reason string) Outcome { + return Outcome{Code: s3_lifecycle_pb.LifecycleDeleteOutcome_SKIPPED_OBJECT_LOCK, Reason: reason} +} + +// FakeLifecycleServer implements s3_lifecycle_pb.SeaweedS3LifecycleInternalServer. +// It returns per-key queued outcomes (FIFO) and falls back to Default when a +// key has no queued entry. Every received request is recorded; tests assert +// against Recorded() in arrival order. +// +// A non-nil Err short-circuits everything — LifecycleDelete returns (nil, Err) +// immediately, before the per-key lookup or the request recording. Use it to +// simulate transport failures. +// +// All methods are safe for concurrent use. Outcomes/Default may be set at +// construction or via Queue/SetDefault between calls; mid-call mutation is +// supported but ordering across that boundary is undefined. +type FakeLifecycleServer struct { + s3_lifecycle_pb.UnimplementedSeaweedS3LifecycleInternalServer + + mu sync.Mutex + queues map[requestKey][]Outcome + def Outcome + err error + received []*s3_lifecycle_pb.LifecycleDeleteRequest +} + +// requestKey is the map key for queues. A struct rather than a delimited +// string so bucket/object/versionId values containing "/" or "@" can't +// collide. +type requestKey struct { + bucket, objectPath, versionId string +} + +// NewFakeLifecycleServer returns a server whose Default outcome is DONE. +// Most tests want a different default; call SetDefault to change it. +func NewFakeLifecycleServer() *FakeLifecycleServer { + return &FakeLifecycleServer{ + queues: map[requestKey][]Outcome{}, + def: Done(), + } +} + +// Queue appends an outcome to the FIFO for (bucket, objectPath, versionId). +// Subsequent calls matching the same key return outcomes in the order they +// were queued; once the queue is drained, Default applies. +func (f *FakeLifecycleServer) Queue(bucket, objectPath, versionId string, outcome Outcome) { + f.mu.Lock() + defer f.mu.Unlock() + k := key(bucket, objectPath, versionId) + f.queues[k] = append(f.queues[k], outcome) +} + +// SetDefault sets the outcome returned when no per-key queue entry remains. +func (f *FakeLifecycleServer) SetDefault(o Outcome) { + f.mu.Lock() + defer f.mu.Unlock() + f.def = o +} + +// SetError makes LifecycleDelete return (nil, err) on every call until +// SetError(nil) clears it. The request is not recorded while Err is set — +// transport-error tests should rely on the worker's own bookkeeping. +func (f *FakeLifecycleServer) SetError(err error) { + f.mu.Lock() + defer f.mu.Unlock() + f.err = err +} + +// Recorded returns a deep-copied snapshot of every request the server has +// received (excluding calls that returned a transport error). Both the +// slice and each element are independent of the fake's internal state, so +// callers can mutate freely without affecting later Recorded() snapshots. +func (f *FakeLifecycleServer) Recorded() []*s3_lifecycle_pb.LifecycleDeleteRequest { + f.mu.Lock() + defer f.mu.Unlock() + out := make([]*s3_lifecycle_pb.LifecycleDeleteRequest, len(f.received)) + for i, r := range f.received { + out[i] = cloneRequest(r) + } + return out +} + +// LifecycleDelete is the gRPC handler. It honors Err first, then dequeues +// the per-key outcome, falling back to Default. +func (f *FakeLifecycleServer) LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) { + f.mu.Lock() + if f.err != nil { + err := f.err + f.mu.Unlock() + return nil, err + } + // Record a deep copy so a caller mutating a Recorded() entry can't + // reach back into the fake's bookkeeping (and so subsequent calls' + // view of "what arrived" stays stable across assertions). + f.received = append(f.received, cloneRequest(req)) + if req == nil { + out := f.def + f.mu.Unlock() + return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: out.Code, Reason: out.Reason}, nil + } + k := key(req.Bucket, req.ObjectPath, req.VersionId) + q := f.queues[k] + var out Outcome + if len(q) > 0 { + out = q[0] + f.queues[k] = q[1:] + } else { + out = f.def + } + f.mu.Unlock() + return &s3_lifecycle_pb.LifecycleDeleteResponse{Outcome: out.Code, Reason: out.Reason}, nil +} + +func key(bucket, objectPath, versionId string) requestKey { + return requestKey{bucket: bucket, objectPath: objectPath, versionId: versionId} +} + +func cloneRequest(req *s3_lifecycle_pb.LifecycleDeleteRequest) *s3_lifecycle_pb.LifecycleDeleteRequest { + if req == nil { + return nil + } + return proto.Clone(req).(*s3_lifecycle_pb.LifecycleDeleteRequest) +} diff --git a/weed/s3api/s3lifecycle/lifecycletest/fakeserver_test.go b/weed/s3api/s3lifecycle/lifecycletest/fakeserver_test.go new file mode 100644 index 000000000..ca9f2439f --- /dev/null +++ b/weed/s3api/s3lifecycle/lifecycletest/fakeserver_test.go @@ -0,0 +1,209 @@ +package lifecycletest + +import ( + "context" + "errors" + "sync" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestFake_DefaultIsDoneOutOfTheBox(t *testing.T) { + // A test that doesn't queue anything should still get a non-error + // response so it can exercise the worker's happy path. + f := NewFakeLifecycleServer() + resp, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{ + Bucket: "b", + ObjectPath: "k", + }) + require.NoError(t, err) + require.NotNil(t, resp) + assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_DONE, resp.Outcome) + assert.Equal(t, "", resp.Reason) +} + +func TestFake_QueuedOutcomesPopFIFO(t *testing.T) { + // Per-key queue is FIFO and one-shot per entry; after the queue + // drains, Default kicks in. + f := NewFakeLifecycleServer() + f.SetDefault(NoopResolved("nothing more queued")) + f.Queue("b", "k", "", RetryLater("first")) + f.Queue("b", "k", "", Blocked("second")) + + got := []s3_lifecycle_pb.LifecycleDeleteOutcome{} + reasons := []string{} + for i := 0; i < 3; i++ { + resp, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{ + Bucket: "b", ObjectPath: "k", + }) + require.NoError(t, err) + got = append(got, resp.Outcome) + reasons = append(reasons, resp.Reason) + } + assert.Equal(t, []s3_lifecycle_pb.LifecycleDeleteOutcome{ + s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER, + s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED, + s3_lifecycle_pb.LifecycleDeleteOutcome_NOOP_RESOLVED, + }, got) + assert.Equal(t, []string{"first", "second", "nothing more queued"}, reasons) +} + +func TestFake_QueuesIsolatedByKey(t *testing.T) { + // Queues are partitioned by (bucket, objectPath, versionId); a queued + // outcome for one key must not bleed into another's lookup. + f := NewFakeLifecycleServer() + f.Queue("b", "a", "", Blocked("a-only")) + f.Queue("b", "b", "", RetryLater("b-only")) + + respA, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "a"}) + require.NoError(t, err) + assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED, respA.Outcome) + + respB, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "b"}) + require.NoError(t, err) + assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_RETRY_LATER, respB.Outcome) +} + +func TestFake_VersionIDPartOfKey(t *testing.T) { + // Two requests for the same bucket/objectPath but different + // versionIds must address different queues. + f := NewFakeLifecycleServer() + f.Queue("b", "k", "v1", SkippedObjectLock("v1-locked")) + f.Queue("b", "k", "v2", Done()) + + respV1, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k", VersionId: "v1"}) + require.NoError(t, err) + respV2, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k", VersionId: "v2"}) + require.NoError(t, err) + assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_SKIPPED_OBJECT_LOCK, respV1.Outcome) + assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_DONE, respV2.Outcome) +} + +func TestFake_KeyComponentsWithDelimitersDoNotCollide(t *testing.T) { + // String-concatenation keys would have made these two requests + // indistinguishable. The struct key keeps them separate. + f := NewFakeLifecycleServer() + f.Queue("b/k", "", "", Blocked("variant-a")) + f.Queue("b", "k", "", Done()) + + respA, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b/k", ObjectPath: ""}) + require.NoError(t, err) + respB, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k"}) + require.NoError(t, err) + assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED, respA.Outcome) + assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_DONE, respB.Outcome) +} + +func TestFake_ErrShortCircuitsBeforeRecording(t *testing.T) { + // Err makes LifecycleDelete return (nil, err) without recording the + // request — transport-error tests rely on the worker's own + // bookkeeping, not the fake's. + f := NewFakeLifecycleServer() + transportErr := errors.New("connection refused") + f.SetError(transportErr) + + resp, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k"}) + assert.Nil(t, resp) + assert.ErrorIs(t, err, transportErr) + assert.Empty(t, f.Recorded(), "transport-error calls must not be recorded") + + // Clearing the error returns the server to normal behavior. + f.SetError(nil) + resp2, err2 := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k"}) + require.NoError(t, err2) + require.NotNil(t, resp2) + assert.Len(t, f.Recorded(), 1) +} + +func TestFake_RecordsRequestsInOrder(t *testing.T) { + // Recorded() preserves arrival order so tests can assert that + // dispatch happened in the expected sequence. + f := NewFakeLifecycleServer() + for _, key := range []string{"k1", "k2", "k3"} { + _, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{ + Bucket: "b", ObjectPath: key, + }) + require.NoError(t, err) + } + rec := f.Recorded() + require.Len(t, rec, 3) + assert.Equal(t, "k1", rec[0].ObjectPath) + assert.Equal(t, "k2", rec[1].ObjectPath) + assert.Equal(t, "k3", rec[2].ObjectPath) +} + +func TestFake_RecordedIsSnapshot(t *testing.T) { + // Mutating the slice the caller got back must not bleed into the + // fake's internal state — otherwise a flaky test could corrupt + // bookkeeping for later assertions. + f := NewFakeLifecycleServer() + _, err := f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k"}) + require.NoError(t, err) + + snap := f.Recorded() + require.Len(t, snap, 1) + snap[0] = nil + again := f.Recorded() + require.Len(t, again, 1) + assert.NotNil(t, again[0], "internal record must survive caller-side mutation of the snapshot") +} + +func TestFake_RecordedRequestsAreDeepCopies(t *testing.T) { + // A caller mutating fields on a Recorded() entry must not bleed + // into a later Recorded() snapshot. We also confirm the original + // request the caller passed in stays decoupled from internal state + // — proto.Clone runs at record time, so the caller's pointer is + // no longer the one the fake holds. + f := NewFakeLifecycleServer() + orig := &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k", VersionId: "v"} + _, err := f.LifecycleDelete(context.Background(), orig) + require.NoError(t, err) + + snap := f.Recorded() + require.Len(t, snap, 1) + snap[0].Bucket = "mutated-by-caller" + snap[0].ObjectPath = "also-mutated" + + again := f.Recorded() + require.Len(t, again, 1) + assert.Equal(t, "b", again[0].Bucket, "field mutations on a snapshot must not bleed back") + assert.Equal(t, "k", again[0].ObjectPath) + assert.Equal(t, "v", again[0].VersionId) + + // The caller's own pointer is independent of the fake's record. + orig.Bucket = "caller-mutates-original" + yetAgain := f.Recorded() + assert.Equal(t, "b", yetAgain[0].Bucket, "caller mutating the input pointer must not bleed in either") +} + +func TestFake_NilRequestUsesDefault(t *testing.T) { + // gRPC won't deliver a nil request in practice, but defensive code + // in the fake should still produce a deterministic response so a + // regressing client doesn't panic the test process. + f := NewFakeLifecycleServer() + f.SetDefault(Blocked("no request")) + resp, err := f.LifecycleDelete(context.Background(), nil) + require.NoError(t, err) + require.NotNil(t, resp) + assert.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED, resp.Outcome) +} + +func TestFake_ConcurrentCallsSerializeWithoutDeadlock(t *testing.T) { + // The dispatcher fans dispatch across many goroutines; the fake + // must not livelock or drop records under concurrent load. + f := NewFakeLifecycleServer() + const N = 64 + var wg sync.WaitGroup + wg.Add(N) + for i := 0; i < N; i++ { + go func() { + defer wg.Done() + _, _ = f.LifecycleDelete(context.Background(), &s3_lifecycle_pb.LifecycleDeleteRequest{Bucket: "b", ObjectPath: "k"}) + }() + } + wg.Wait() + assert.Len(t, f.Recorded(), N) +}