diff --git a/weed/credential/filer_etc/filer_etc_group.go b/weed/credential/filer_etc/filer_etc_group.go index 1a381fb93..36a5d4fb4 100644 --- a/weed/credential/filer_etc/filer_etc_group.go +++ b/weed/credential/filer_etc/filer_etc_group.go @@ -78,7 +78,7 @@ func (store *FilerEtcStore) saveGroup(ctx context.Context, group *iam_pb.Group) if err != nil { return err } - return filer.SaveInsideFiler(client, filer.IamConfigDirectory+"/"+IamGroupsDirectory, group.Name+".json", data) + return filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory+"/"+IamGroupsDirectory, group.Name+".json", data) }) } diff --git a/weed/credential/filer_etc/filer_etc_identity.go b/weed/credential/filer_etc/filer_etc_identity.go index 9c4900ab1..f3b093381 100644 --- a/weed/credential/filer_etc/filer_etc_identity.go +++ b/weed/credential/filer_etc/filer_etc_identity.go @@ -484,7 +484,7 @@ func (store *FilerEtcStore) saveIdentity(ctx context.Context, identity *iam_pb.I if err != nil { return err } - return filer.SaveInsideFiler(client, filer.IamConfigDirectory+"/"+IamIdentitiesDirectory, identity.Name+".json", data) + return filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory+"/"+IamIdentitiesDirectory, identity.Name+".json", data) }) } diff --git a/weed/credential/filer_etc/filer_etc_policy.go b/weed/credential/filer_etc/filer_etc_policy.go index 2c632ea42..b7231bd8f 100644 --- a/weed/credential/filer_etc/filer_etc_policy.go +++ b/weed/credential/filer_etc/filer_etc_policy.go @@ -64,7 +64,7 @@ func (store *FilerEtcStore) saveLegacyPoliciesCollection(ctx context.Context, po if err != nil { return err } - return filer.SaveInsideFiler(client, filer.IamConfigDirectory, filer.IamPoliciesFile, content) + return filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory, filer.IamPoliciesFile, content) }) } @@ -216,7 +216,7 @@ func (store *FilerEtcStore) savePolicy(ctx context.Context, name string, documen if err != nil { return err } - return filer.SaveInsideFiler(client, filer.IamConfigDirectory+"/"+IamPoliciesDirectory, name+".json", data) + return filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory+"/"+IamPoliciesDirectory, name+".json", data) }) } diff --git a/weed/credential/filer_etc/filer_etc_service_account.go b/weed/credential/filer_etc/filer_etc_service_account.go index 94324b90b..2f8b3a480 100644 --- a/weed/credential/filer_etc/filer_etc_service_account.go +++ b/weed/credential/filer_etc/filer_etc_service_account.go @@ -71,7 +71,7 @@ func (store *FilerEtcStore) saveServiceAccount(ctx context.Context, sa *iam_pb.S if err != nil { return err } - return filer.SaveInsideFiler(client, filer.IamConfigDirectory+"/"+IamServiceAccountsDirectory, sa.Id+".json", data) + return filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory+"/"+IamServiceAccountsDirectory, sa.Id+".json", data) }) } diff --git a/weed/filer/read_write.go b/weed/filer/read_write.go index 976f88da1..1288e0649 100644 --- a/weed/filer/read_write.go +++ b/weed/filer/read_write.go @@ -41,15 +41,15 @@ func ReadInsideFiler(ctx context.Context, filerClient filer_pb.SeaweedFilerClien return } -func SaveInsideFiler(client filer_pb.SeaweedFilerClient, dir, name string, content []byte) error { +func SaveInsideFiler(ctx context.Context, client filer_pb.SeaweedFilerClient, dir, name string, content []byte) error { - resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{ + resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{ Directory: dir, Name: name, }) if err == filer_pb.ErrNotFound { - err = filer_pb.CreateEntry(context.Background(), client, &filer_pb.CreateEntryRequest{ + err = filer_pb.CreateEntry(ctx, client, &filer_pb.CreateEntryRequest{ Directory: dir, Entry: &filer_pb.Entry{ Name: name, @@ -69,7 +69,7 @@ func SaveInsideFiler(client filer_pb.SeaweedFilerClient, dir, name string, conte entry.Content = content entry.Attributes.Mtime = time.Now().Unix() entry.Attributes.FileSize = uint64(len(content)) - err = filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{ + err = filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{ Directory: dir, Entry: entry, }) diff --git a/weed/filer/remote_mapping.go b/weed/filer/remote_mapping.go index bfb609d6b..9c488d92e 100644 --- a/weed/filer/remote_mapping.go +++ b/weed/filer/remote_mapping.go @@ -52,7 +52,7 @@ func InsertMountMapping(filerClient filer_pb.FilerClient, dir string, remoteStor // save back err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return SaveInsideFiler(client, DirectoryEtcRemote, REMOTE_STORAGE_MOUNT_FILE, newContent) + return SaveInsideFiler(context.Background(), client, DirectoryEtcRemote, REMOTE_STORAGE_MOUNT_FILE, newContent) }) if err != nil { return fmt.Errorf("save mapping: %w", err) @@ -83,7 +83,7 @@ func DeleteMountMapping(filerClient filer_pb.FilerClient, dir string) (err error // save back err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return SaveInsideFiler(client, DirectoryEtcRemote, REMOTE_STORAGE_MOUNT_FILE, newContent) + return SaveInsideFiler(context.Background(), client, DirectoryEtcRemote, REMOTE_STORAGE_MOUNT_FILE, newContent) }) if err != nil { return fmt.Errorf("save mapping: %w", err) diff --git a/weed/iamapi/iamapi_server.go b/weed/iamapi/iamapi_server.go index 26eaf289f..34d9e612f 100644 --- a/weed/iamapi/iamapi_server.go +++ b/weed/iamapi/iamapi_server.go @@ -193,7 +193,7 @@ func (iama *IamS3ApiConfigure) PutS3ApiConfigurationToFiler(s3cfg *iam_pb.S3ApiC } return pb.WithOneOfGrpcFilerClients(false, iama.option.Filers, iama.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error { err = util.Retry("saveIamIdentity", func() error { - return filer.SaveInsideFiler(client, filer.IamConfigDirectory, filer.IamIdentityFile, buf.Bytes()) + return filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory, filer.IamIdentityFile, buf.Bytes()) }) if err != nil { return err @@ -229,7 +229,7 @@ func (iama *IamS3ApiConfigure) PutPolicies(policies *Policies) (err error) { return err } return pb.WithOneOfGrpcFilerClients(false, iama.option.Filers, iama.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error { - if err := filer.SaveInsideFiler(client, filer.IamConfigDirectory, filer.IamPoliciesFile, b); err != nil { + if err := filer.SaveInsideFiler(context.Background(), client, filer.IamConfigDirectory, filer.IamPoliciesFile, b); err != nil { return err } return nil diff --git a/weed/mq/kafka/gateway/coordinator_registry.go b/weed/mq/kafka/gateway/coordinator_registry.go index 6e651a5d8..152157463 100644 --- a/weed/mq/kafka/gateway/coordinator_registry.go +++ b/weed/mq/kafka/gateway/coordinator_registry.go @@ -662,7 +662,7 @@ func (cr *CoordinatorRegistry) saveCoordinatorAssignment(consumerGroup string, a // Save to individual file: /topics/kafka/.meta/coordinators/_assignments.json fileName := fmt.Sprintf("%s_assignments.json", consumerGroup) - return filer.SaveInsideFiler(client, CoordinatorAssignmentsDir, fileName, assignmentData) + return filer.SaveInsideFiler(context.Background(), client, CoordinatorAssignmentsDir, fileName, assignmentData) }) } diff --git a/weed/mq/offset/consumer_group_storage.go b/weed/mq/offset/consumer_group_storage.go index c38e77308..13721c707 100644 --- a/weed/mq/offset/consumer_group_storage.go +++ b/weed/mq/offset/consumer_group_storage.go @@ -86,7 +86,7 @@ func (f *FilerConsumerGroupOffsetStorage) SaveConsumerGroupPosition(t topic.Topi } return f.filerClientAccessor.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return filer.SaveInsideFiler(client, consumersDir, offsetFileName, jsonBytes) + return filer.SaveInsideFiler(context.Background(), client, consumersDir, offsetFileName, jsonBytes) }) } diff --git a/weed/mq/offset/filer_storage.go b/weed/mq/offset/filer_storage.go index 10b54bae3..6b86fc7fd 100644 --- a/weed/mq/offset/filer_storage.go +++ b/weed/mq/offset/filer_storage.go @@ -38,7 +38,7 @@ func (f *FilerOffsetStorage) SaveCheckpoint(namespace, topicName string, partiti util.Uint64toBytes(offsetBytes, uint64(offset)) return f.filerClientAccessor.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return filer.SaveInsideFiler(client, partitionDir, fileName, offsetBytes) + return filer.SaveInsideFiler(context.Background(), client, partitionDir, fileName, offsetBytes) }) } diff --git a/weed/mq/topic/topic.go b/weed/mq/topic/topic.go index ad8c4864d..11a2fe808 100644 --- a/weed/mq/topic/topic.go +++ b/weed/mq/topic/topic.go @@ -100,7 +100,7 @@ func (t Topic) ReadConfFileWithMetadata(client filer_pb.SeaweedFilerClient) (*mq func (t Topic) WriteConfFile(client filer_pb.SeaweedFilerClient, conf *mq_pb.ConfigureTopicResponse) error { var buf bytes.Buffer filer.ProtoToText(&buf, conf) - if err := filer.SaveInsideFiler(client, t.Dir(), filer.TopicConfFile, buf.Bytes()); err != nil { + if err := filer.SaveInsideFiler(context.Background(), client, t.Dir(), filer.TopicConfFile, buf.Bytes()); err != nil { return fmt.Errorf("save topic %v conf: %w", t, err) } return nil diff --git a/weed/s3api/s3api_bucket_handlers.go b/weed/s3api/s3api_bucket_handlers.go index ef24b40bf..bbabb0db5 100644 --- a/weed/s3api/s3api_bucket_handlers.go +++ b/weed/s3api/s3api_bucket_handlers.go @@ -1051,7 +1051,7 @@ func (s3a *S3ApiServer) PutBucketLifecycleConfigurationHandler(w http.ResponseWr s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) } if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return filer.SaveInsideFiler(client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes()) + return filer.SaveInsideFiler(context.Background(), client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes()) }); err != nil { glog.Errorf("PutBucketLifecycleConfigurationHandler save config inside filer: %s", err) s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) @@ -1106,7 +1106,7 @@ func (s3a *S3ApiServer) DeleteBucketLifecycleHandler(w http.ResponseWriter, r *h s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) } if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return filer.SaveInsideFiler(client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes()) + return filer.SaveInsideFiler(context.Background(), client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes()) }); err != nil { glog.Errorf("DeleteBucketLifecycleHandler save config inside filer: %s", err) s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) diff --git a/weed/s3api/s3lifecycle/dispatcher/blocker.go b/weed/s3api/s3lifecycle/dispatcher/blocker.go deleted file mode 100644 index c189353db..000000000 --- a/weed/s3api/s3lifecycle/dispatcher/blocker.go +++ /dev/null @@ -1,77 +0,0 @@ -package dispatcher - -import ( - "context" - "sync" - "time" - - "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" -) - -// BlockerRecord persists the state needed to re-freeze a cursor on worker -// restart. Stored under /etc/s3/lifecycle/blockers///. -// -// Operator action via the blocker-resolve flow either: -// - Quarantine: keep the freeze; mark the (object, version) skipped via a -// side-table so subsequent re-evaluations don't trip the same blocker. -// - Retry: clear the record and Unfreeze; the cursor advances on the next -// successful dispatch. -type BlockerRecord struct { - ShardID int - Key s3lifecycle.ActionKey - FrozenAtNs int64 // tsNs of the event that tripped the blocker - Reason string // RPC outcome reason - CreatedAt time.Time -} - -// BlockerStore persists BlockerRecords. The contract: -// - Put replaces any existing record for (ShardID, Key) atomically. -// - Delete is idempotent (no error on missing). -// - List returns all records for ShardID for restart-time freeze replay. -type BlockerStore interface { - Put(ctx context.Context, rec BlockerRecord) error - Delete(ctx context.Context, shardID int, key s3lifecycle.ActionKey) error - List(ctx context.Context, shardID int) ([]BlockerRecord, error) -} - -// InMemoryBlockerStore is a BlockerStore for tests. -type InMemoryBlockerStore struct { - mu sync.Mutex - records map[int]map[s3lifecycle.ActionKey]BlockerRecord -} - -func NewInMemoryBlockerStore() *InMemoryBlockerStore { - return &InMemoryBlockerStore{records: map[int]map[s3lifecycle.ActionKey]BlockerRecord{}} -} - -func (s *InMemoryBlockerStore) Put(ctx context.Context, rec BlockerRecord) error { - s.mu.Lock() - defer s.mu.Unlock() - m, ok := s.records[rec.ShardID] - if !ok { - m = map[s3lifecycle.ActionKey]BlockerRecord{} - s.records[rec.ShardID] = m - } - m[rec.Key] = rec - return nil -} - -func (s *InMemoryBlockerStore) Delete(ctx context.Context, shardID int, key s3lifecycle.ActionKey) error { - s.mu.Lock() - defer s.mu.Unlock() - if m, ok := s.records[shardID]; ok { - delete(m, key) - } - return nil -} - -func (s *InMemoryBlockerStore) List(ctx context.Context, shardID int) ([]BlockerRecord, error) { - s.mu.Lock() - defer s.mu.Unlock() - m := s.records[shardID] - out := make([]BlockerRecord, 0, len(m)) - for _, rec := range m { - out = append(out, rec) - } - return out, nil -} diff --git a/weed/s3api/s3lifecycle/dispatcher/dispatcher.go b/weed/s3api/s3lifecycle/dispatcher/dispatcher.go index 2e7015d86..6de626b9b 100644 --- a/weed/s3api/s3lifecycle/dispatcher/dispatcher.go +++ b/weed/s3api/s3lifecycle/dispatcher/dispatcher.go @@ -20,19 +20,22 @@ type LifecycleClient interface { } // Dispatcher consumes due Matches, calls LifecycleDelete, and routes the -// outcome back to the per-shard cursor and the blocker store. +// outcome back to the per-shard cursor. // // State machine for one Match: // DONE / NOOP_RESOLVED / SKIPPED_OBJECT_LOCK -> Cursor.Advance // RETRY_LATER (within budget) -> back into the schedule with backoff -// RETRY_LATER (budget exhausted) -> escalate to BLOCKED -// BLOCKED -> BlockerStore.Put + Cursor.Freeze -// FATAL_EVENT_ERROR / unknown -> treat as BLOCKED +// RETRY_LATER (budget exhausted) / BLOCKED -> Cursor.Freeze in-memory +// FATAL_EVENT_ERROR / unknown -> treat as BLOCKED +// +// A frozen cursor doesn't advance, so the durable cursor is the durable +// "stuck" state on its own: a worker restart re-encounters the poison +// event at MinTsNs and re-freezes after the same retry cycle. No separate +// blocker store is needed. type Dispatcher struct { ShardID int Client LifecycleClient Cursor *reader.Cursor - Blockers BlockerStore Schedule *router.Schedule // RetryBudget caps RETRY_LATER attempts before escalating to BLOCKED. @@ -44,8 +47,8 @@ type Dispatcher struct { RetryBackoff time.Duration // retries[Match.Key+ObjectKey] = attempts so far. In-memory only: - // worker restart resets the budget, which is fine because BLOCKED is - // durable and the cursor is durable. + // worker restart resets the budget, which is fine because the cursor + // is durable and the same poison event will land us here again. retries map[retryKey]int } @@ -169,33 +172,11 @@ func (d *Dispatcher) handleRetryLater(ctx context.Context, m router.Match, reaso func (d *Dispatcher) handleBlocked(ctx context.Context, m router.Match, reason string) { delete(d.retries, keyOf(m)) - rec := BlockerRecord{ - ShardID: d.ShardID, - Key: m.Key, - FrozenAtNs: m.EventTs.UnixNano(), - Reason: reason, - CreatedAt: time.Now(), - } - if err := d.Blockers.Put(ctx, rec); err != nil { - // Persistence failure: retry; cursor stays frozen in-memory. - glog.Errorf("lifecycle blocker persist: shard=%d key=%v: %v", d.ShardID, m.Key, err) - } + glog.Warningf("lifecycle: cursor frozen shard=%d key=%+v eventTs=%s reason=%s", + d.ShardID, m.Key, m.EventTs.UTC().Format(time.RFC3339Nano), reason) d.Cursor.Freeze(m.Key, m.EventTs.UnixNano()) } -// ReplayBlockers re-applies in-memory freezes from the durable BlockerStore. -// Call once on Pipeline startup before the reader begins emitting. -func (d *Dispatcher) ReplayBlockers(ctx context.Context) error { - recs, err := d.Blockers.List(ctx, d.ShardID) - if err != nil { - return fmt.Errorf("blocker list: %w", err) - } - for _, r := range recs { - d.Cursor.Freeze(r.Key, r.FrozenAtNs) - } - return nil -} - func toProtoActionKind(k s3lifecycle.ActionKind) s3_lifecycle_pb.ActionKind { switch k { case s3lifecycle.ActionKindExpirationDays: diff --git a/weed/s3api/s3lifecycle/dispatcher/dispatcher_test.go b/weed/s3api/s3lifecycle/dispatcher/dispatcher_test.go index 90de082cf..9b900647a 100644 --- a/weed/s3api/s3lifecycle/dispatcher/dispatcher_test.go +++ b/weed/s3api/s3lifecycle/dispatcher/dispatcher_test.go @@ -38,7 +38,6 @@ func newDispatcher(client LifecycleClient) (*Dispatcher, *router.Schedule) { ShardID: 0, Client: client, Cursor: reader.NewCursor(), - Blockers: NewInMemoryBlockerStore(), Schedule: sched, RetryBudget: 3, RetryBackoff: time.Millisecond, @@ -142,12 +141,8 @@ func TestDispatchRetryBudgetEscalatesToBlocked(t *testing.T) { if !d.Cursor.IsFrozen(m.Key) { t.Fatal("expected freeze after budget exhausted") } - recs, _ := d.Blockers.List(context.Background(), 0) - if len(recs) != 1 { - t.Fatalf("expected 1 blocker record, got %d", len(recs)) - } - if recs[0].Reason == "" { - t.Fatal("blocker record missing reason") + if d.Cursor.Get(m.Key) != t0.UnixNano() { + t.Fatal("frozen cursor should be pinned at event ts") } } @@ -241,27 +236,33 @@ func TestDispatchSkipsFrozenCursor(t *testing.T) { } } -func TestReplayBlockersRefreezes(t *testing.T) { - store := NewInMemoryBlockerStore() +func TestDispatchRestartReFreezesNaturally(t *testing.T) { + // No durable blocker store: the durable cursor + a deterministic poison + // event self-recover to the blocked state on a fresh Dispatcher. After + // the budget burns, the new cursor freezes at the same EventTs. + respond := func(int) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) { + return &s3_lifecycle_pb.LifecycleDeleteResponse{ + Outcome: s3_lifecycle_pb.LifecycleDeleteOutcome_BLOCKED, + Reason: "deterministic poison", + }, nil + } + d, sched := newDispatcher(&fakeClient{respond: respond}) t0 := time.Now() - key := s3lifecycle.ActionKey{Bucket: "bk", ActionKind: s3lifecycle.ActionKindExpirationDays} - store.Put(context.Background(), BlockerRecord{ - ShardID: 0, - Key: key, - FrozenAtNs: t0.UnixNano(), - Reason: "prior run", - CreatedAt: t0, - }) - d := &Dispatcher{ - ShardID: 0, - Cursor: reader.NewCursor(), - Blockers: store, - Schedule: router.NewSchedule(), + m := mkMatch(t0, t0, "obj") + sched.Add(m) + d.Tick(context.Background(), t0) + if !d.Cursor.IsFrozen(m.Key) { + t.Fatal("first run should freeze") } - if err := d.ReplayBlockers(context.Background()); err != nil { - t.Fatalf("ReplayBlockers: %v", err) + + // Simulate restart: brand-new Dispatcher and Cursor, same poison event. + d2, sched2 := newDispatcher(&fakeClient{respond: respond}) + sched2.Add(m) + d2.Tick(context.Background(), t0) + if !d2.Cursor.IsFrozen(m.Key) { + t.Fatal("restart should re-freeze without a durable blocker store") } - if !d.Cursor.IsFrozen(key) { - t.Fatal("ReplayBlockers should refreeze cursor") + if d2.Cursor.Get(m.Key) != t0.UnixNano() { + t.Fatal("re-freeze cursor not pinned at event ts") } } diff --git a/weed/s3api/s3lifecycle/dispatcher/filer_persister.go b/weed/s3api/s3lifecycle/dispatcher/filer_persister.go new file mode 100644 index 000000000..679ca1fd9 --- /dev/null +++ b/weed/s3api/s3lifecycle/dispatcher/filer_persister.go @@ -0,0 +1,141 @@ +package dispatcher + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "sort" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader" +) + +// FilerStore is the small subset of filer-client operations the persistence +// layer needs. The default implementation calls filer.ReadInsideFiler / +// filer.SaveInsideFiler; tests inject an in-memory fake. +type FilerStore interface { + Read(ctx context.Context, dir, name string) ([]byte, error) + Save(ctx context.Context, dir, name string, content []byte) error +} + +// NewFilerStoreClient adapts a SeaweedFilerClient into a FilerStore. +func NewFilerStoreClient(client filer_pb.SeaweedFilerClient) FilerStore { + return &filerStoreClient{client: client} +} + +type filerStoreClient struct { + client filer_pb.SeaweedFilerClient +} + +func (s *filerStoreClient) Read(ctx context.Context, dir, name string) ([]byte, error) { + return filer.ReadInsideFiler(ctx, s.client, dir, name) +} + +func (s *filerStoreClient) Save(ctx context.Context, dir, name string, content []byte) error { + return filer.SaveInsideFiler(ctx, s.client, dir, name, content) +} + +// CursorDir is the filer directory holding per-shard cursor files. +const CursorDir = "/etc/s3/lifecycle/cursors" + +// FilerPersister persists per-shard cursor maps to /etc/s3/lifecycle/cursors/ +// as JSON. One file per shard keeps Save atomic — the filer writes the entry +// in a single mutation, so a crash mid-write doesn't leak partial state. +type FilerPersister struct { + Store FilerStore +} + +// cursorFile is the on-disk JSON shape. cursorFileEntry repeats the +// ActionKey fields explicitly so the format stays human-readable and stable +// against Go-side struct rearrangements. +type cursorFile struct { + Version int `json:"version"` + ShardID int `json:"shard_id"` + Entries []cursorFileEntry `json:"entries"` +} + +type cursorFileEntry struct { + Bucket string `json:"bucket"` + RuleHash []byte `json:"rule_hash"` // base64 in JSON + ActionKind int `json:"action_kind"` + TsNs int64 `json:"ts_ns"` +} + +const cursorFileVersion = 1 + +func cursorFileName(shardID int) string { + return fmt.Sprintf("shard-%02d.json", shardID) +} + +func (p *FilerPersister) Load(ctx context.Context, shardID int) (map[s3lifecycle.ActionKey]int64, error) { + if p.Store == nil { + return nil, errors.New("FilerPersister: nil Store") + } + content, err := p.Store.Read(ctx, CursorDir, cursorFileName(shardID)) + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) { + return map[s3lifecycle.ActionKey]int64{}, nil + } + return nil, fmt.Errorf("cursor read shard=%d: %w", shardID, err) + } + if len(content) == 0 { + return map[s3lifecycle.ActionKey]int64{}, nil + } + var cf cursorFile + if err := json.Unmarshal(content, &cf); err != nil { + return nil, fmt.Errorf("cursor decode shard=%d: %w", shardID, err) + } + out := make(map[s3lifecycle.ActionKey]int64, len(cf.Entries)) + for _, e := range cf.Entries { + k := s3lifecycle.ActionKey{ + Bucket: e.Bucket, + ActionKind: s3lifecycle.ActionKind(e.ActionKind), + } + copy(k.RuleHash[:], e.RuleHash) + out[k] = e.TsNs + } + return out, nil +} + +func (p *FilerPersister) Save(ctx context.Context, shardID int, state map[s3lifecycle.ActionKey]int64) error { + if p.Store == nil { + return errors.New("FilerPersister: nil Store") + } + cf := cursorFile{Version: cursorFileVersion, ShardID: shardID} + cf.Entries = make([]cursorFileEntry, 0, len(state)) + for k, v := range state { + hash := k.RuleHash + cf.Entries = append(cf.Entries, cursorFileEntry{ + Bucket: k.Bucket, + RuleHash: hash[:], + ActionKind: int(k.ActionKind), + TsNs: v, + }) + } + // Stable order so the on-disk file diffs cleanly across saves. + sort.Slice(cf.Entries, func(i, j int) bool { + a, b := cf.Entries[i], cf.Entries[j] + if a.Bucket != b.Bucket { + return a.Bucket < b.Bucket + } + if a.ActionKind != b.ActionKind { + return a.ActionKind < b.ActionKind + } + return bytes.Compare(a.RuleHash, b.RuleHash) < 0 + }) + var buf bytes.Buffer + if err := json.NewEncoder(&buf).Encode(cf); err != nil { + return fmt.Errorf("cursor encode shard=%d: %w", shardID, err) + } + if err := p.Store.Save(ctx, CursorDir, cursorFileName(shardID), buf.Bytes()); err != nil { + return fmt.Errorf("cursor save shard=%d: %w", shardID, err) + } + return nil +} + +// Compile-time interface check. +var _ reader.Persister = (*FilerPersister)(nil) diff --git a/weed/s3api/s3lifecycle/dispatcher/filer_persister_test.go b/weed/s3api/s3lifecycle/dispatcher/filer_persister_test.go new file mode 100644 index 000000000..638a8cf77 --- /dev/null +++ b/weed/s3api/s3lifecycle/dispatcher/filer_persister_test.go @@ -0,0 +1,159 @@ +package dispatcher + +import ( + "context" + "sync" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" +) + +// fakeFilerStore is an in-memory FilerStore for tests. +type fakeFilerStore struct { + mu sync.Mutex + files map[string][]byte +} + +func newFakeFilerStore() *fakeFilerStore { + return &fakeFilerStore{files: map[string][]byte{}} +} + +func (f *fakeFilerStore) key(dir, name string) string { return dir + "/" + name } + +func (f *fakeFilerStore) Read(ctx context.Context, dir, name string) ([]byte, error) { + f.mu.Lock() + defer f.mu.Unlock() + v, ok := f.files[f.key(dir, name)] + if !ok { + return nil, filer_pb.ErrNotFound + } + return append([]byte(nil), v...), nil +} + +func (f *fakeFilerStore) Save(ctx context.Context, dir, name string, content []byte) error { + f.mu.Lock() + defer f.mu.Unlock() + f.files[f.key(dir, name)] = append([]byte(nil), content...) + return nil +} + +func mkKey(bucket string, kind s3lifecycle.ActionKind, hashByte byte) s3lifecycle.ActionKey { + k := s3lifecycle.ActionKey{Bucket: bucket, ActionKind: kind} + for i := range k.RuleHash { + k.RuleHash[i] = hashByte + } + return k +} + +func TestFilerPersisterEmptyLoadReturnsEmptyMap(t *testing.T) { + p := &FilerPersister{Store: newFakeFilerStore()} + state, err := p.Load(context.Background(), 0) + if err != nil { + t.Fatalf("Load on empty: %v", err) + } + if len(state) != 0 { + t.Fatalf("expected empty map, got %d entries", len(state)) + } +} + +func TestFilerPersisterSaveLoadRoundTrip(t *testing.T) { + p := &FilerPersister{Store: newFakeFilerStore()} + ctx := context.Background() + + in := map[s3lifecycle.ActionKey]int64{ + mkKey("bucket-a", s3lifecycle.ActionKindExpirationDays, 0xAA): 12345, + mkKey("bucket-b", s3lifecycle.ActionKindAbortMPU, 0xBB): 67890, + mkKey("bucket-a", s3lifecycle.ActionKindNoncurrentDays, 0xCC): 54321, + } + if err := p.Save(ctx, 3, in); err != nil { + t.Fatalf("Save: %v", err) + } + out, err := p.Load(ctx, 3) + if err != nil { + t.Fatalf("Load: %v", err) + } + if len(out) != len(in) { + t.Fatalf("Load count=%d, want %d", len(out), len(in)) + } + for k, v := range in { + if got := out[k]; got != v { + t.Fatalf("Load[%v]=%d, want %d", k, got, v) + } + } +} + +func TestFilerPersisterIsolatesShards(t *testing.T) { + p := &FilerPersister{Store: newFakeFilerStore()} + ctx := context.Background() + + stateA := map[s3lifecycle.ActionKey]int64{mkKey("a", s3lifecycle.ActionKindExpirationDays, 1): 100} + stateB := map[s3lifecycle.ActionKey]int64{mkKey("a", s3lifecycle.ActionKindExpirationDays, 1): 200} + if err := p.Save(ctx, 0, stateA); err != nil { + t.Fatalf("Save 0: %v", err) + } + if err := p.Save(ctx, 1, stateB); err != nil { + t.Fatalf("Save 1: %v", err) + } + loadA, _ := p.Load(ctx, 0) + loadB, _ := p.Load(ctx, 1) + if loadA[mkKey("a", s3lifecycle.ActionKindExpirationDays, 1)] != 100 { + t.Fatalf("shard 0 leaked from shard 1: %v", loadA) + } + if loadB[mkKey("a", s3lifecycle.ActionKindExpirationDays, 1)] != 200 { + t.Fatalf("shard 1 reads stale: %v", loadB) + } +} + +func TestFilerPersisterSaveOverwrites(t *testing.T) { + p := &FilerPersister{Store: newFakeFilerStore()} + ctx := context.Background() + k := mkKey("b", s3lifecycle.ActionKindExpirationDays, 0xAA) + + if err := p.Save(ctx, 0, map[s3lifecycle.ActionKey]int64{k: 100}); err != nil { + t.Fatalf("Save 1: %v", err) + } + if err := p.Save(ctx, 0, map[s3lifecycle.ActionKey]int64{k: 200}); err != nil { + t.Fatalf("Save 2: %v", err) + } + out, _ := p.Load(ctx, 0) + if out[k] != 200 { + t.Fatalf("overwrite not applied, got %d", out[k]) + } +} + +func TestFilerPersisterSaveIsDeterministic(t *testing.T) { + // Saving the same map twice must produce byte-identical content so the + // on-disk file diffs cleanly when the state hasn't changed. + store := newFakeFilerStore() + p := &FilerPersister{Store: store} + ctx := context.Background() + + in := map[s3lifecycle.ActionKey]int64{ + mkKey("zeta", s3lifecycle.ActionKindNoncurrentDays, 0xCC): 300, + mkKey("alpha", s3lifecycle.ActionKindExpirationDays, 0xAA): 100, + mkKey("alpha", s3lifecycle.ActionKindAbortMPU, 0xBB): 200, + } + if err := p.Save(ctx, 0, in); err != nil { + t.Fatalf("Save 1: %v", err) + } + first := append([]byte(nil), store.files[store.key(CursorDir, cursorFileName(0))]...) + + if err := p.Save(ctx, 0, in); err != nil { + t.Fatalf("Save 2: %v", err) + } + second := store.files[store.key(CursorDir, cursorFileName(0))] + if string(first) != string(second) { + t.Fatalf("non-deterministic save:\n first=%s\nsecond=%s", first, second) + } +} + +func TestFilerPersisterCorruptDataReturnsError(t *testing.T) { + store := newFakeFilerStore() + store.Save(context.Background(), CursorDir, "shard-00.json", []byte("not json")) + + p := &FilerPersister{Store: store} + if _, err := p.Load(context.Background(), 0); err == nil { + t.Fatal("expected decode error, got nil") + } +} diff --git a/weed/s3api/s3lifecycle/dispatcher/pipeline.go b/weed/s3api/s3lifecycle/dispatcher/pipeline.go index d7735c42f..c376af4a5 100644 --- a/weed/s3api/s3lifecycle/dispatcher/pipeline.go +++ b/weed/s3api/s3lifecycle/dispatcher/pipeline.go @@ -23,7 +23,6 @@ type Pipeline struct { Engine *engine.Engine Cursor *reader.Cursor Persister reader.Persister - Blockers BlockerStore Client LifecycleClient FilerClient filer_pb.SeaweedFilerClient @@ -55,14 +54,16 @@ const ( // is the durable buffer, so a restart re-derives them). func (p *Pipeline) Run(ctx context.Context) error { if p.Engine == nil || p.Cursor == nil || p.Persister == nil || - p.Blockers == nil || p.Client == nil || p.FilerClient == nil { + p.Client == nil || p.FilerClient == nil { return errors.New("pipeline: missing required dependency") } if p.BucketsPath == "" { return errors.New("pipeline: BucketsPath required") } - // 1. Restore cursor + replay blocker freezes. + // Restore cursor; freezes re-arm naturally when the reader re-encounters + // the poison event at MinTsNs and the dispatch state machine drives it + // back to BLOCKED. state, err := p.Persister.Load(ctx, p.ShardID) if err != nil { return fmt.Errorf("cursor load: %w", err) @@ -73,12 +74,8 @@ func (p *Pipeline) Run(ctx context.Context) error { ShardID: p.ShardID, Client: p.Client, Cursor: p.Cursor, - Blockers: p.Blockers, Schedule: router.NewSchedule(), } - if err := dispatch.ReplayBlockers(ctx); err != nil { - return fmt.Errorf("blocker replay: %w", err) - } // 2. Wire reader -> router -> schedule via a buffered channel. bufSize := p.EventBuffer diff --git a/weed/shell/command_fs_configure.go b/weed/shell/command_fs_configure.go index 84568c766..fac4ed17e 100644 --- a/weed/shell/command_fs_configure.go +++ b/weed/shell/command_fs_configure.go @@ -2,6 +2,7 @@ package shell import ( "bytes" + "context" "flag" "fmt" "io" @@ -142,7 +143,7 @@ func (c *commandFsConfigure) Do(args []string, commandEnv *CommandEnv, writer io if *apply { if err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return filer.SaveInsideFiler(client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf2.Bytes()) + return filer.SaveInsideFiler(context.Background(), client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf2.Bytes()) }); err != nil && err != filer_pb.ErrNotFound { return err } diff --git a/weed/shell/command_remote_configure.go b/weed/shell/command_remote_configure.go index 66dc21ea6..09fce04f7 100644 --- a/weed/shell/command_remote_configure.go +++ b/weed/shell/command_remote_configure.go @@ -205,7 +205,7 @@ func (c *commandRemoteConfigure) saveRemoteStorage(commandEnv *CommandEnv, write } if err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return filer.SaveInsideFiler(client, filer.DirectoryEtcRemote, conf.Name+filer.REMOTE_STORAGE_CONF_SUFFIX, data) + return filer.SaveInsideFiler(context.Background(), client, filer.DirectoryEtcRemote, conf.Name+filer.REMOTE_STORAGE_CONF_SUFFIX, data) }); err != nil && err != filer_pb.ErrNotFound { return err } diff --git a/weed/shell/command_s3_bucket_quota_check.go b/weed/shell/command_s3_bucket_quota_check.go index c92b52117..100c70f17 100644 --- a/weed/shell/command_s3_bucket_quota_check.go +++ b/weed/shell/command_s3_bucket_quota_check.go @@ -92,7 +92,7 @@ func (c *commandS3BucketQuotaEnforce) Do(args []string, commandEnv *CommandEnv, fc.ToText(&buf2) if err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return filer.SaveInsideFiler(client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf2.Bytes()) + return filer.SaveInsideFiler(context.Background(), client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf2.Bytes()) }); err != nil && err != filer_pb.ErrNotFound { return err } diff --git a/weed/shell/command_s3_circuitbreaker.go b/weed/shell/command_s3_circuitbreaker.go index 43ff454f7..772cacc78 100644 --- a/weed/shell/command_s3_circuitbreaker.go +++ b/weed/shell/command_s3_circuitbreaker.go @@ -2,6 +2,7 @@ package shell import ( "bytes" + "context" "flag" "fmt" "io" @@ -186,7 +187,7 @@ func (c *commandS3CircuitBreaker) Do(args []string, commandEnv *CommandEnv, writ if *apply { if err := commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return filer.SaveInsideFiler(client, dir, file, buf.Bytes()) + return filer.SaveInsideFiler(context.Background(), client, dir, file, buf.Bytes()) }); err != nil { return err }