Files
s3/weed/mount/meta_cache/meta_cache_subscribe.go
zooqueen dd74d3e6d6 Replace AGPL MinIO fork with Apache-2.0 SeaweedFS fork; rebrand CLI weed->s3
hanzoai/s3 was an AGPL-3.0 MinIO fork whose network-copyleft blocks embedding
and reselling as a managed service. Replace it wholesale with a fork of
SeaweedFS (github.com/seaweedfs/seaweedfs, Apache-2.0, Go) which ships a
built-in S3 gateway + Filer.

- License: AGPL-3.0 -> Apache-2.0 (upstream LICENSE kept verbatim, incl.
  "Copyright 2025 Chris Lu"); NOTICE added crediting SeaweedFS. No AGPL/BUSL/
  SSPL/GPL or enterprise ("ee") files remain.
- Module: go.mod path set to github.com/hanzoai/s3 with a self-replace
  (replace github.com/seaweedfs/seaweedfs => ./) so the ~1800 internal Go files
  keep their upstream import paths with zero churn. Internal identifiers and
  import paths are intentionally NOT renamed.
- CLI rebrand: the binary is built as `s3` (was `weed`); root usage, per-command
  help/usage/flag strings, shell autocomplete program name, and the product name
  in CLI output now read "Hanzo S3" / `s3`. Users never see `weed`.
  (e.g. `s3 server`, `s3 filer`, `s3 master`, `s3 volume`, `s3 s3`, `s3 mount`.)
- Build: `make` (or `go build -o s3 ./weed`) builds the `s3` binary green;
  `cmd/weed-db` and `cmd/weed-sql` also build; `go vet ./weed/command/` clean.
- README/LLM.md rewritten for Hanzo S3; AGENTS.md/CLAUDE.md symlinks preserved.
- Dropped a 33MB prebuilt Mach-O `stress` binary upstream had committed; `.gitignore`
  updated to ignore the `s3` build output.
2026-06-18 23:41:25 -07:00

102 lines
3.2 KiB
Go

package meta_cache
import (
"context"
"io"
"strings"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/util"
)
type MetadataFollower struct {
PathPrefixToWatch string
ProcessEventFn func(resp *filer_pb.SubscribeMetadataResponse) error
}
func mergeProcessors(mainProcessor func(resp *filer_pb.SubscribeMetadataResponse) error, followers ...*MetadataFollower) func(resp *filer_pb.SubscribeMetadataResponse) error {
return func(resp *filer_pb.SubscribeMetadataResponse) error {
// build the full path
entry := resp.EventNotification.NewEntry
if entry == nil {
entry = resp.EventNotification.OldEntry
}
if entry != nil {
dir := resp.Directory
if resp.EventNotification.NewParentPath != "" {
dir = resp.EventNotification.NewParentPath
}
fp := util.NewFullPath(dir, entry.Name)
for _, follower := range followers {
if strings.HasPrefix(string(fp), follower.PathPrefixToWatch) {
if err := follower.ProcessEventFn(resp); err != nil {
return err
}
}
}
}
return mainProcessor(resp)
}
}
func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.FilerClient, dir string, lastTsNs int64, skipSelfEvents bool, onRetry func(lastTsNs int64, err error), followers ...*MetadataFollower) error {
var prefixes []string
for _, follower := range followers {
prefixes = append(prefixes, follower.PathPrefixToWatch)
}
processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
if skipSelfEvents && resp.EventNotification != nil {
for _, sig := range resp.EventNotification.Signatures {
if sig == selfSignature {
glog.V(4).Infof("skip self-originated event %s", resp.Directory)
return nil
}
}
}
return mc.ApplyMetadataResponse(context.Background(), resp, SubscriberMetadataResponseApplyOptions)
}
prefix := dir
if !strings.HasSuffix(prefix, "/") {
prefix = prefix + "/"
}
// Read persisted log chunks directly from volume servers, keeping the replay
// cost off the filer's heap (see LogFileReaderFn below).
lookupFn := filer.LookupFn(client)
metadataFollowOption := &pb.MetadataFollowOption{
ClientName: "mount",
ClientId: selfSignature,
ClientEpoch: 1,
SelfSignature: selfSignature,
PathPrefix: prefix,
AdditionalPathPrefixes: prefixes,
DirectoriesToWatch: nil,
StartTsNs: lastTsNs,
StopTsNs: 0,
EventErrorType: pb.FatalOnError,
LogFileReaderFn: func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) {
return filer.NewChunkStreamReaderFromLookup(context.Background(), lookupFn, chunks), nil
},
}
util.RetryUntil("followMetaUpdates", func() error {
metadataFollowOption.ClientEpoch++
return pb.WithFilerClientFollowMetadata(client, metadataFollowOption, mergeProcessors(processEventFn, followers...))
}, func(err error) bool {
if onRetry != nil {
onRetry(metadataFollowOption.StartTsNs, err)
}
glog.Errorf("follow metadata updates: %v", err)
return true
})
return nil
}