eric/distributor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

Centralize storage walk emission

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-02 18:50:50 +00:00
parent c372a02357
commit d5e3aa7a44
5 changed files with 185 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
internal/adapters/local/backend.go
View File

@@ -161,31 +161,10 @@ func (b *Backend) Walk(ctx context.Context, prefix string, opts storage.WalkOpti
}
return b.translateError(storage.OpWalk, prefix, err)
}
visited := 0
emit := func(entry storage.Entry) error {
if err := ctx.Err(); err != nil {
return err
}
if opts.Limit > 0 && visited >= opts.Limit {
return storage.ErrStopWalk
}
visited++
if err := fn(entry); err != nil {
if errors.Is(err, storage.ErrStopWalk) {
return storage.ErrStopWalk
}
return storage.NewError(storage.OpWalk, backendName, entry.Path, storage.ErrUnknown, err)
}
return nil
}
emitter := storage.NewWalkEmitter(ctx, backendName, opts, fn)
if !info.IsDir() {
if err := emit(entryFromInfo(prefix, info)); errors.Is(err, storage.ErrStopWalk) {
return nil
} else if err != nil {
return err
}
return nil
return storage.FinishWalk(emitter.Emit(entryFromInfo(prefix, info)))
}
walkErr := filepath.WalkDir(nativePrefix, func(nativePath string, dirEntry fs.DirEntry, err error) error {
@@ -210,24 +189,13 @@ func (b *Backend) Walk(ctx context.Context, prefix string, opts storage.WalkOpti
if err != nil {
return b.translateError(storage.OpWalk, logicalPath, err)
}
return emit(entryFromInfo(logicalPath, info))
return emitter.Emit(entryFromInfo(logicalPath, info))
})
if errors.Is(walkErr, storage.ErrStopWalk) {
return nil
}
return walkErr
return storage.FinishWalk(walkErr)
}
func (b *Backend) HasAny(ctx context.Context, prefix string) (bool, error) {
found := false
err := b.Walk(ctx, prefix, storage.WalkOptions{Recursive: false, Limit: 1}, func(storage.Entry) error {
found = true
return storage.ErrStopWalk
})
if err != nil {
return false, err
}
return found, nil
return storage.HasAny(ctx, b, prefix)
}
func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error {

42
internal/adapters/s3/backend.go
View File

@@ -183,55 +183,25 @@ func (b *Backend) Walk(ctx context.Context, logicalPrefix string, opts storage.W
if err := storage.ValidatePrefix(logicalPrefix); err != nil {
return err
}
visited := 0
emit := func(entry storage.Entry) error {
if err := ctx.Err(); err != nil {
return err
}
if opts.Limit > 0 && visited >= opts.Limit {
return storage.ErrStopWalk
}
visited++
if err := fn(entry); err != nil {
if errors.Is(err, storage.ErrStopWalk) {
return storage.ErrStopWalk
}
return storage.NewError(storage.OpWalk, BackendName, entry.Path, storage.ErrUnknown, err)
}
return nil
}
emitter := storage.NewWalkEmitter(ctx, BackendName, opts, fn)
if logicalPrefix != "" {
entry, err := b.Stat(ctx, logicalPrefix)
if err == nil {
if err := emit(entry); errors.Is(err, storage.ErrStopWalk) {
return nil
} else if err != nil {
return err
if err := emitter.Emit(entry); err != nil {
return storage.FinishWalk(err)
}
if opts.Limit > 0 && visited >= opts.Limit {
if emitter.LimitReached() {
return nil
}
} else if !storage.IsNotFound(err) {
return err
}
}
err := b.walkObjects(ctx, logicalPrefix, opts, emit)
if errors.Is(err, storage.ErrStopWalk) {
return nil
}
return err
return storage.FinishWalk(b.walkObjects(ctx, logicalPrefix, opts, emitter.Emit))
}
func (b *Backend) HasAny(ctx context.Context, logicalPrefix string) (bool, error) {
found := false
err := b.Walk(ctx, logicalPrefix, storage.WalkOptions{Recursive: false, Limit: 1}, func(storage.Entry) error {
found = true
return storage.ErrStopWalk
})
if err != nil {
return false, err
}
return found, nil
return storage.HasAny(ctx, b, logicalPrefix)
}
func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error {

42
internal/adapters/ssh/backend.go
View File

@@ -209,51 +209,17 @@ func (b *Backend) Walk(ctx context.Context, prefix string, opts storage.WalkOpti
return b.translateError(storage.OpWalk, prefix, err)
}
visited := 0
emit := func(entry storage.Entry) error {
if err := ctx.Err(); err != nil {
return err
}
if opts.Limit > 0 && visited >= opts.Limit {
return storage.ErrStopWalk
}
visited++
if err := fn(entry); err != nil {
if errors.Is(err, storage.ErrStopWalk) {
return storage.ErrStopWalk
}
return storage.NewError(storage.OpWalk, BackendName, entry.Path, storage.ErrUnknown, err)
}
return nil
}
emitter := storage.NewWalkEmitter(ctx, BackendName, opts, fn)
if !info.IsDir() {
if err := emit(entryFromInfo(prefix, info)); errors.Is(err, storage.ErrStopWalk) {
return nil
} else if err != nil {
return err
}
return nil
return storage.FinishWalk(emitter.Emit(entryFromInfo(prefix, info)))
}
if err := b.walkDirectory(ctx, prefix, nativePrefix, opts, emit); errors.Is(err, storage.ErrStopWalk) {
return nil
} else if err != nil {
return err
}
return nil
return storage.FinishWalk(b.walkDirectory(ctx, prefix, nativePrefix, opts, emitter.Emit))
}
func (b *Backend) HasAny(ctx context.Context, prefix string) (bool, error) {
found := false
err := b.Walk(ctx, prefix, storage.WalkOptions{Recursive: false, Limit: 1}, func(storage.Entry) error {
found = true
return storage.ErrStopWalk
})
if err != nil {
return false, err
}
return found, nil
return storage.HasAny(ctx, b, prefix)
}
func (b *Backend) DeleteManagedBundle(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error {