Files
localai-org-maint-botandEttore Di Giacinto 8b5f62cc02 fix(distributed): bound ephemeral staging (#11924)
* docs: design ephemeral staging retention

High-frequency camera and audio inputs can fill a worker before the current six-hour cleanup window expires.

Define a one-hour retention policy that preserves recently modified request payloads.

Assisted-by: Codex:gpt-6

* docs: make ephemeral staging request-owned

Time-based retention can still fill a worker under bursty or high-rate input. Define request-lifecycle cleanup with capacity reservation and crash recovery.

Assisted-by: Codex:gpt-6

* feat(distributed): release exact staged keys

Request inputs need transport-neutral cleanup after backend calls. Add authenticated exact-key deletion for HTTP and coordinated cache eviction before shared-object deletion for S3/NATS.

Preserve URL metacharacters as filename data, reject unsafe keys, and remove upload sidecars while pruning empty request directories.

Assisted-by: Codex:gpt-6

* fix(distributed): release staged request inputs

Ephemeral inputs remained on workers after inference completed. Release each exact key after synchronous and streaming calls, including partial staging failures.

Use a bounded cleanup context so caller cancellation cannot suppress release. Preserve caller requests and backend results when cleanup fails.

Assisted-by: Codex:gpt-6

* feat(worker): bound ephemeral staging capacity

Concurrent staging can otherwise exceed its byte limit or consume reserved filesystem headroom. Explicit states keep bytes charged through each reservation, write, and commit transition.

Use a synchronized waiter count to prove Commit blocks until bounded writers close, and retain committed baselines across re-reservation.

Assisted-by: Codex:gpt-6

* feat(worker): enforce ephemeral staging bounds

Share capacity accounting across HTTP and S3 request inputs so workers
reject uploads before exhausting their filesystem. Reconcile exact release
and crash recovery with the same guard.

Assisted-by: Codex:gpt-6

* fix(distributed): make staged release race-safe

Pin each release path component before removing request-owned inputs and sidecars. Stop pruning when a directory identity changes.

Assisted-by: Codex:gpt-6

* fix(worker): retain staged input ownership

Keep committed request inputs protected from age recovery until exact release ends their ownership. Startup-scanned files remain reclaimable and can acquire ownership through reservation.

Assisted-by: Codex:gpt-6

* fix(worker): claim cached ephemeral inputs

Keep startup-scanned cache hits owned while inference uses them and reconcile their actual size against capacity.

Assisted-by: Codex:gpt-6

* fix(distributed): enforce staging admission

Propagate multimodal staging failures before inference and claim matching ephemeral HTTP cache entries. Fall back to PUT when an older worker does not support claims.

Assisted-by: Codex:gpt-6

* fix(distributed): close staging accounting gaps

Keep unknown-length reservations charged until bytes reach disk and bound NATS release waits by the lifecycle cleanup deadline.

Assisted-by: Codex:gpt-6

* fix(distributed): restage swept cache hits

Treat files removed between cache probing and ownership claims as misses so HTTP and S3 workers can stage them again.

Assisted-by: Codex:gpt-6

* fix(distributed): release staged inputs by request

Release every input from one inference with one fixed-size worker coordination request. Fence request ingress against cleanup, bound staging capacity and cleanup state, and retain exact-key release for rolling upgrades.

Assisted-by: Codex:gpt-6

---------

Co-authored-by: Ettore Di Giacinto <mudler@localai.io>
2026-09-08 18:13:40 +02:00

145 lines
4.7 KiB
Go

//go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
package safefile
import (
"errors"
"fmt"
"path/filepath"
"strings"
"golang.org/x/sys/unix"
)
// ErrUnsafePath reports a path shape or file type that exact removal refuses.
var ErrUnsafePath = errors.New("unsafe removal path")
// RemoveExact removes a file and its named sidecars below root without
// following symbolic links. It prunes up to pruneParents empty parent
// directories, but never removes root itself.
func RemoveExact(root, relativePath string, sidecarSuffixes []string, pruneParents int) error {
return removeExact(root, relativePath, sidecarSuffixes, pruneParents, nil)
}
func removeExact(root, relativePath string, sidecarSuffixes []string, pruneParents int, parentsOpened func()) error {
parts, err := cleanRelativeParts(relativePath)
if err != nil {
return err
}
if pruneParents < 0 {
return fmt.Errorf("%w: prune parent count must not be negative", ErrUnsafePath)
}
rootFD, err := unix.Open(root, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0)
if err != nil {
return fmt.Errorf("opening removal root %q: %w", root, err)
}
handles := []int{rootFD}
defer func() {
for i := len(handles) - 1; i >= 0; i-- {
_ = unix.Close(handles[i])
}
}()
for _, component := range parts[:len(parts)-1] {
fd, openErr := unix.Openat(handles[len(handles)-1], component, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0)
if errors.Is(openErr, unix.ENOENT) {
return nil
}
if openErr != nil {
if errors.Is(openErr, unix.ELOOP) || errors.Is(openErr, unix.ENOTDIR) {
return fmt.Errorf("%w: path component %q is not a directory: %v", ErrUnsafePath, component, openErr)
}
return fmt.Errorf("opening removal path component %q: %w", component, openErr)
}
handles = append(handles, fd)
}
if parentsOpened != nil {
parentsOpened()
}
parentFD := handles[len(handles)-1]
leaf := parts[len(parts)-1]
names := make([]string, 0, len(sidecarSuffixes)+1)
names = append(names, leaf)
for _, suffix := range sidecarSuffixes {
if suffix == "" || strings.ContainsAny(suffix, `/\\`) {
return fmt.Errorf("%w: invalid sidecar suffix %q", ErrUnsafePath, suffix)
}
names = append(names, leaf+suffix)
}
for _, name := range names {
var stat unix.Stat_t
statErr := unix.Fstatat(parentFD, name, &stat, unix.AT_SYMLINK_NOFOLLOW)
if errors.Is(statErr, unix.ENOENT) {
continue
}
if statErr != nil {
return fmt.Errorf("stating removal entry %q: %w", name, statErr)
}
switch stat.Mode & unix.S_IFMT {
case unix.S_IFLNK:
return fmt.Errorf("%w: refusing to remove symbolic link %q", ErrUnsafePath, name)
case unix.S_IFDIR:
return fmt.Errorf("%w: release key identifies a directory", ErrUnsafePath)
}
}
for _, name := range names {
if unlinkErr := unix.Unlinkat(parentFD, name, 0); unlinkErr != nil && !errors.Is(unlinkErr, unix.ENOENT) {
return fmt.Errorf("removing entry %q: %w", name, unlinkErr)
}
}
maxPrune := min(pruneParents, len(handles)-1)
for childIndex := len(handles) - 1; childIndex >= len(handles)-maxPrune; childIndex-- {
parentIndex := childIndex - 1
name := parts[childIndex-1]
same, identityErr := sameDirectoryEntry(handles[parentIndex], name, handles[childIndex])
if identityErr != nil {
if errors.Is(identityErr, unix.ENOENT) {
break
}
return fmt.Errorf("checking directory %q before pruning: %w", name, identityErr)
}
if !same {
break
}
removeErr := unix.Unlinkat(handles[parentIndex], name, unix.AT_REMOVEDIR)
if removeErr == nil {
continue
}
if errors.Is(removeErr, unix.ENOENT) || errors.Is(removeErr, unix.ENOTEMPTY) || errors.Is(removeErr, unix.EEXIST) {
break
}
return fmt.Errorf("pruning directory %q: %w", name, removeErr)
}
return nil
}
func cleanRelativeParts(relativePath string) ([]string, error) {
if relativePath == "" || filepath.IsAbs(relativePath) || filepath.Clean(relativePath) != relativePath {
return nil, fmt.Errorf("%w: %q is not a clean relative path", ErrUnsafePath, relativePath)
}
parts := strings.Split(relativePath, string(filepath.Separator))
for _, part := range parts {
if part == "" || part == "." || part == ".." {
return nil, fmt.Errorf("%w: %q is not a clean relative path", ErrUnsafePath, relativePath)
}
}
return parts, nil
}
func sameDirectoryEntry(parentFD int, name string, openedFD int) (bool, error) {
var opened unix.Stat_t
if err := unix.Fstat(openedFD, &opened); err != nil {
return false, err
}
var current unix.Stat_t
if err := unix.Fstatat(parentFD, name, &current, unix.AT_SYMLINK_NOFOLLOW); err != nil {
return false, err
}
return current.Mode&unix.S_IFMT == unix.S_IFDIR && current.Dev == opened.Dev && current.Ino == opened.Ino, nil
}