Compare commits
3 Commits
b60264701c
...
0886662360
| Author | SHA1 | Date | |
|---|---|---|---|
| 0886662360 | |||
| 439ceb5d8e | |||
| ecaf3407fa |
+102
-85
@@ -38,6 +38,12 @@ const (
|
||||
DefaultMaxDescription = 400
|
||||
)
|
||||
|
||||
// The two transports, as Status reports them to the web surface.
|
||||
const (
|
||||
transportStdio = "stdio"
|
||||
transportHTTP = "http"
|
||||
)
|
||||
|
||||
var (
|
||||
// ErrNoServer — the named server is not configured.
|
||||
ErrNoServer = errors.New("mcp: no such server")
|
||||
@@ -90,6 +96,12 @@ type ServerConfig struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
}
|
||||
|
||||
// argv is the stdio server's command line. It is argv and never a shell
|
||||
// string, so the same slice serves both the exec and the /tools display.
|
||||
func (c ServerConfig) argv() []string {
|
||||
return append([]string{c.Command}, c.Args...)
|
||||
}
|
||||
|
||||
// PosterFactory builds the HTTP door for one server. It is a factory rather
|
||||
// than a single shared Poster because allow_private is per server: the fetcher
|
||||
// that may reach http://localhost:9100/mcp must NOT be the same fetcher another
|
||||
@@ -230,18 +242,7 @@ func (m *Manager) dial(ctx context.Context, name string) error {
|
||||
c.lastTry = time.Now()
|
||||
m.mu.Unlock()
|
||||
|
||||
var tr transport
|
||||
var err error
|
||||
if cfg.Command != "" {
|
||||
tr, err = newStdioTransport(ctx, append([]string{cfg.Command}, cfg.Args...), cfg.Env, cfg.Dir)
|
||||
} else if m.newPoster == nil {
|
||||
err = fmt.Errorf("server %q has a url but no http door was wired", name)
|
||||
} else {
|
||||
var poster Poster
|
||||
if poster, err = m.newPoster(cfg); err == nil {
|
||||
tr = newHTTPTransport(poster, cfg.URL, cfg.Headers)
|
||||
}
|
||||
}
|
||||
tr, err := m.openTransport(ctx, cfg)
|
||||
if err != nil {
|
||||
m.fail(name, err)
|
||||
return err
|
||||
@@ -262,21 +263,44 @@ func (m *Manager) dial(ctx context.Context, name string) error {
|
||||
tools = nil
|
||||
}
|
||||
tools = filterTools(cfg, tools)
|
||||
|
||||
m.mu.Lock()
|
||||
if old := m.conns[name].client; old != nil {
|
||||
_ = old.Close()
|
||||
}
|
||||
m.conns[name].client = cl
|
||||
m.conns[name].tools = tools
|
||||
m.conns[name].lastErr = nil
|
||||
m.conns[name].fails = 0
|
||||
m.conns[name].dialedAt = time.Now()
|
||||
m.mu.Unlock()
|
||||
m.succeed(name, cl, tools)
|
||||
log.Printf("mcp: %s connected (%s %s), %d tool(s)", name, cl.Info().Name, cl.Info().Version, len(tools))
|
||||
return nil
|
||||
}
|
||||
|
||||
// openTransport builds the door this server is configured for. A url server
|
||||
// with no factory is one server's problem, reported here, so a bad block never
|
||||
// stops the daemon.
|
||||
func (m *Manager) openTransport(ctx context.Context, cfg ServerConfig) (transport, error) {
|
||||
if cfg.Command != "" {
|
||||
return newStdioTransport(ctx, cfg.argv(), cfg.Env, cfg.Dir)
|
||||
}
|
||||
if m.newPoster == nil {
|
||||
return nil, fmt.Errorf("server %q has a url but no http door was wired", cfg.Name)
|
||||
}
|
||||
poster, err := m.newPoster(cfg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return newHTTPTransport(poster, cfg.URL, cfg.Headers), nil
|
||||
}
|
||||
|
||||
// succeed records a live connection and closes the one it replaces, so a
|
||||
// re-dial does not leak the previous subprocess.
|
||||
func (m *Manager) succeed(name string, cl *Client, tools []Tool) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
c := m.conns[name]
|
||||
if c == nil {
|
||||
return
|
||||
}
|
||||
if c.client != nil {
|
||||
_ = c.client.Close()
|
||||
}
|
||||
c.client, c.tools, c.lastErr, c.fails = cl, tools, nil, 0
|
||||
c.dialedAt = time.Now()
|
||||
}
|
||||
|
||||
func (m *Manager) fail(name string, err error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
@@ -288,8 +312,7 @@ func (m *Manager) fail(name string, err error) {
|
||||
}
|
||||
}
|
||||
|
||||
// filterTools applies AllowTools and MaxTools, drops nameless entries and
|
||||
// truncates descriptions.
|
||||
// filterTools applies AllowTools and MaxTools and truncates descriptions.
|
||||
//
|
||||
// Over the cap WITHOUT allow_tools, the whole contribution is dropped. Taking
|
||||
// the first N of a sorted list was deterministic but it handed the choice of
|
||||
@@ -423,9 +446,9 @@ func (m *Manager) Status() []Status {
|
||||
c := m.conns[name]
|
||||
s := Status{Name: name, Tools: len(c.tools)}
|
||||
if c.cfg.Command != "" {
|
||||
s.Transport, s.Target = "stdio", strings.Join(append([]string{c.cfg.Command}, c.cfg.Args...), " ")
|
||||
s.Transport, s.Target = transportStdio, strings.Join(c.cfg.argv(), " ")
|
||||
} else {
|
||||
s.Transport, s.Target = "http", c.cfg.URL
|
||||
s.Transport, s.Target = transportHTTP, c.cfg.URL
|
||||
}
|
||||
if c.client != nil {
|
||||
s.Connected = true
|
||||
@@ -442,21 +465,10 @@ func (m *Manager) Status() []Status {
|
||||
// Call runs server's tool with args. Args come from the router and nothing
|
||||
// else; there is no path here through which a note or a fact could travel.
|
||||
func (m *Manager) Call(ctx context.Context, server, tool string, args map[string]any) (string, error) {
|
||||
m.mu.Lock()
|
||||
c := m.conns[server]
|
||||
m.mu.Unlock()
|
||||
if c == nil {
|
||||
cl, cfg, _, configured, known := m.lookup(server, tool)
|
||||
if !configured {
|
||||
return "", fmt.Errorf("%w: %s", ErrNoServer, server)
|
||||
}
|
||||
m.mu.Lock()
|
||||
cl, timeout, known := c.client, c.cfg.Timeout, false
|
||||
for _, t := range c.tools {
|
||||
if t.Name == tool {
|
||||
known = true
|
||||
break
|
||||
}
|
||||
}
|
||||
m.mu.Unlock()
|
||||
if cl == nil {
|
||||
return "", fmt.Errorf("%w: %s", ErrNotConnected, server)
|
||||
}
|
||||
@@ -466,11 +478,30 @@ func (m *Manager) Call(ctx context.Context, server, tool string, args map[string
|
||||
if !known {
|
||||
return "", fmt.Errorf("%w: %s/%s", ErrToolGone, server, tool)
|
||||
}
|
||||
cctx, cancel := context.WithTimeout(ctx, timeout)
|
||||
cctx, cancel := context.WithTimeout(ctx, cfg.Timeout)
|
||||
defer cancel()
|
||||
return cl.CallTool(cctx, tool, args)
|
||||
}
|
||||
|
||||
// lookup reads one server's live state under the lock. Every call path needs
|
||||
// the same four answers, and reading them in one critical section keeps a
|
||||
// server that goes down mid-check from answering half yes.
|
||||
func (m *Manager) lookup(server, tool string) (cl *Client, cfg ServerConfig, found Tool, configured, known bool) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
c := m.conns[server]
|
||||
if c == nil {
|
||||
return nil, ServerConfig{}, Tool{}, false, false
|
||||
}
|
||||
for _, t := range c.tools {
|
||||
if t.Name == tool {
|
||||
found, known = t, true
|
||||
break
|
||||
}
|
||||
}
|
||||
return c.client, c.cfg, found, true, known
|
||||
}
|
||||
|
||||
// Resources lists resources across connected servers.
|
||||
func (m *Manager) Resources(ctx context.Context) []Resource {
|
||||
m.mu.Lock()
|
||||
@@ -494,18 +525,11 @@ func (m *Manager) Resources(ctx context.Context) []Resource {
|
||||
|
||||
// ReadResource reads one resource from one server.
|
||||
func (m *Manager) ReadResource(ctx context.Context, server, uri string) (string, error) {
|
||||
m.mu.Lock()
|
||||
c := m.conns[server]
|
||||
var cl *Client
|
||||
var timeout time.Duration
|
||||
if c != nil {
|
||||
cl, timeout = c.client, c.cfg.Timeout
|
||||
}
|
||||
m.mu.Unlock()
|
||||
cl, cfg, _, _, _ := m.lookup(server, "")
|
||||
if cl == nil {
|
||||
return "", fmt.Errorf("%w: %s", ErrNoServer, server)
|
||||
}
|
||||
cctx, cancel := context.WithTimeout(ctx, timeout)
|
||||
cctx, cancel := context.WithTimeout(ctx, cfg.Timeout)
|
||||
defer cancel()
|
||||
return cl.ReadResource(cctx, uri)
|
||||
}
|
||||
@@ -559,32 +583,18 @@ var ErrNeedsArgs = errors.New("mcp: tool needs named arguments")
|
||||
// router picks tools by name similarity and the description a human reads is
|
||||
// server-written too.
|
||||
func (m *Manager) CallPositional(ctx context.Context, server, tool string, args []string) (string, error) {
|
||||
m.mu.Lock()
|
||||
c := m.conns[server]
|
||||
var schema json.RawMessage
|
||||
found, readOnly, bindable := false, false, false
|
||||
configured, connected := c != nil, false
|
||||
if c != nil {
|
||||
connected = c.client != nil
|
||||
for _, t := range c.tools {
|
||||
if t.Name == tool {
|
||||
schema, readOnly, found = t.InputSchema, t.ReadOnly, true
|
||||
bindable = contains(c.cfg.AllowTools, tool)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
m.mu.Unlock()
|
||||
if !found {
|
||||
cl, cfg, t, configured, known := m.lookup(server, tool)
|
||||
if !known {
|
||||
if !configured {
|
||||
return "", fmt.Errorf("%w: %s", ErrNoServer, server)
|
||||
}
|
||||
if !connected {
|
||||
if cl == nil {
|
||||
return "", fmt.Errorf("%w: %s", ErrNotConnected, server)
|
||||
}
|
||||
return "", fmt.Errorf("%w: %s/%s", ErrToolGone, server, tool)
|
||||
}
|
||||
named, err := bindPositional(schema, args, readOnly && bindable)
|
||||
bindable := t.ReadOnly && contains(cfg.AllowTools, tool)
|
||||
named, err := bindPositional(t.InputSchema, args, bindable)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
@@ -620,23 +630,30 @@ func bindPositional(schema json.RawMessage, args []string, bind bool) (map[strin
|
||||
if !bind {
|
||||
return nil, fmt.Errorf("%w: %q, and a guessed argument goes only to a read-only tool named in allow_tools", ErrNeedsArgs, name)
|
||||
}
|
||||
tail := strings.TrimSpace(strings.Join(args, " "))
|
||||
if tail == "" {
|
||||
return nil, fmt.Errorf("%w: %q", ErrNeedsArgs, name)
|
||||
}
|
||||
switch prop.Type {
|
||||
case "string", "":
|
||||
return map[string]any{name: tail}, nil
|
||||
case "integer", "number":
|
||||
n, err := strconv.ParseFloat(tail, 64)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: %q wants a number, got %q", ErrNeedsArgs, name, tail)
|
||||
}
|
||||
return map[string]any{name: n}, nil
|
||||
default:
|
||||
return nil, fmt.Errorf("%w: %q is a %s", ErrNeedsArgs, name, prop.Type)
|
||||
}
|
||||
return bindOne(name, prop.Type, args)
|
||||
default:
|
||||
return nil, fmt.Errorf("%w: %s", ErrNeedsArgs, strings.Join(s.Required, ", "))
|
||||
}
|
||||
}
|
||||
|
||||
// bindOne puts the whole positional tail in the one required property. The
|
||||
// tail is spoken words, so only a scalar can hold it and anything else is
|
||||
// refused rather than coerced.
|
||||
func bindOne(name, typ string, args []string) (map[string]any, error) {
|
||||
tail := strings.TrimSpace(strings.Join(args, " "))
|
||||
if tail == "" {
|
||||
return nil, fmt.Errorf("%w: %q", ErrNeedsArgs, name)
|
||||
}
|
||||
switch typ {
|
||||
case "string", "":
|
||||
return map[string]any{name: tail}, nil
|
||||
case "integer", "number":
|
||||
n, err := strconv.ParseFloat(tail, 64)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: %q wants a number, got %q", ErrNeedsArgs, name, tail)
|
||||
}
|
||||
return map[string]any{name: n}, nil
|
||||
default:
|
||||
return nil, fmt.Errorf("%w: %q is a %s", ErrNeedsArgs, name, typ)
|
||||
}
|
||||
}
|
||||
|
||||
+146
-104
@@ -42,6 +42,29 @@ const DefaultRetention = 7 * 24 * time.Hour
|
||||
// thousand photos inside the window.
|
||||
const DefaultMaxTotalBytes int64 = 4 << 30
|
||||
|
||||
const (
|
||||
// dirPerm and filePerm: these are recordings of people, so the daemon's
|
||||
// user is the only reader.
|
||||
dirPerm fs.FileMode = 0o700
|
||||
filePerm fs.FileMode = 0o600
|
||||
// metaExt is the sidecar suffix. Anything else under a bucket is blob
|
||||
// bytes, which is how the walkers tell the two apart.
|
||||
metaExt = ".json"
|
||||
// spoolName is the incremental-write directory, held outside the kind
|
||||
// directories so no walker mistakes a half-written file for a blob.
|
||||
spoolName = "spool"
|
||||
// bucketPrefix is how many leading id characters name the subdirectory, so
|
||||
// one kind is spread over 256 directories rather than one flat listing.
|
||||
bucketPrefix = 2
|
||||
// idLen is the length of a hex sha256, which is the only id shape a path
|
||||
// is ever built from.
|
||||
idLen = 64
|
||||
)
|
||||
|
||||
// allKinds is every kind a walker has to visit. A store-wide operation covers
|
||||
// all of them, and this is the one list to extend when a third kind lands.
|
||||
var allKinds = []Kind{KindImage, KindAudio}
|
||||
|
||||
// ErrStoreFull — the store is at its total-bytes budget. Distinct from
|
||||
// ErrTooLarge: the payload is a reasonable size and there is no room for it, so
|
||||
// the answer is to prune or raise the budget, not to send something smaller.
|
||||
@@ -85,7 +108,7 @@ func OpenWithBudget(dir string, maxBytes, maxTotal int64, retention time.Duratio
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("media: resolve dir: %w", err)
|
||||
}
|
||||
if err := os.MkdirAll(abs, 0o700); err != nil {
|
||||
if err := os.MkdirAll(abs, dirPerm); err != nil {
|
||||
return nil, fmt.Errorf("media: create dir: %w", err)
|
||||
}
|
||||
if maxBytes <= 0 {
|
||||
@@ -117,14 +140,14 @@ func OpenWithBudget(dir string, maxBytes, maxTotal int64, retention time.Duratio
|
||||
// over at zero.
|
||||
func (s *Store) measure() int64 {
|
||||
var total int64
|
||||
spool := filepath.Join(s.dir, "spool")
|
||||
spool := filepath.Join(s.dir, spoolName)
|
||||
_ = filepath.WalkDir(s.dir, func(path string, d fs.DirEntry, err error) error {
|
||||
if err == nil && d.IsDir() && path == spool {
|
||||
// Spool files are not blobs yet and PutFile counts them when they
|
||||
// become one. Counting them here too would double them.
|
||||
return filepath.SkipDir
|
||||
}
|
||||
if err != nil || d.IsDir() || strings.HasSuffix(path, ".json") {
|
||||
if err != nil || d.IsDir() || strings.HasSuffix(path, metaExt) {
|
||||
return nil //nolint:nilerr // an unreadable corner is not worth refusing to boot over
|
||||
}
|
||||
if info, err := d.Info(); err == nil {
|
||||
@@ -153,6 +176,54 @@ func (s *Store) Dir() string { return s.dir }
|
||||
// Retention is the configured age limit Prune enforces.
|
||||
func (s *Store) Retention() time.Duration { return s.retention }
|
||||
|
||||
// reserve claims size against the whole-store budget and reports ErrStoreFull
|
||||
// when there is no room. Claiming before the write means two concurrent Puts
|
||||
// cannot both pass a check that only one of them fits through.
|
||||
func (s *Store) reserve(size int64) error {
|
||||
s.totalMu.Lock()
|
||||
room := s.total+size <= s.maxTotal
|
||||
if room {
|
||||
s.total += size
|
||||
}
|
||||
stored := s.total
|
||||
s.totalMu.Unlock()
|
||||
if !room {
|
||||
return fmt.Errorf("%w: %d stored, %d budget, %d more asked for",
|
||||
ErrStoreFull, stored, s.maxTotal, size)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// release gives size back, for a reservation whose write failed and for bytes
|
||||
// a delete removed. The floor at zero keeps a miscount from reading as a store
|
||||
// that owes itself space.
|
||||
func (s *Store) release(size int64) {
|
||||
s.totalMu.Lock()
|
||||
s.total -= size
|
||||
if s.total < 0 {
|
||||
s.total = 0
|
||||
}
|
||||
s.totalMu.Unlock()
|
||||
}
|
||||
|
||||
// bucket is the directory an id lives in. Every path the store builds goes
|
||||
// through here, so the traversal guard in validID has one place to sit.
|
||||
func (s *Store) bucket(kind Kind, id string) string {
|
||||
return filepath.Join(s.dir, string(kind), id[:bucketPrefix])
|
||||
}
|
||||
|
||||
// newBlob describes what is about to be stored. A blob already on disk keeps
|
||||
// its first-seen time: re-sending the same photo every hour must not keep it
|
||||
// alive past retention.
|
||||
func (s *Store) newBlob(kind Kind, mime, source, id, blobPath, metaPath string, size int64) Blob {
|
||||
b := Blob{ID: id, Kind: kind, MIME: mime, Size: size, Source: source,
|
||||
Created: s.now().UTC(), Path: blobPath}
|
||||
if prev, err := readMeta(metaPath); err == nil && !prev.Created.IsZero() {
|
||||
b.Created = prev.Created
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
// Put stores data and returns its Blob. The id is the sha256 of data, so
|
||||
// storing the same bytes twice is idempotent: the second call rewrites the
|
||||
// sidecar (keeping the ORIGINAL creation time, so a re-send cannot extend
|
||||
@@ -178,32 +249,19 @@ func (s *Store) Put(kind Kind, mime, source string, data []byte) (Blob, error) {
|
||||
if err != nil {
|
||||
return Blob{}, err
|
||||
}
|
||||
if err := os.MkdirAll(filepath.Dir(blobPath), 0o700); err != nil {
|
||||
if err := os.MkdirAll(filepath.Dir(blobPath), dirPerm); err != nil {
|
||||
return Blob{}, fmt.Errorf("media: create bucket: %w", err)
|
||||
}
|
||||
|
||||
b := Blob{ID: id, Kind: kind, MIME: mime, Size: int64(len(data)), Source: source,
|
||||
Created: s.now().UTC(), Path: blobPath}
|
||||
|
||||
// A blob already here keeps its first-seen time. Re-sending the same photo
|
||||
// every hour must not keep it alive past retention.
|
||||
if prev, err := readMeta(metaPath); err == nil && !prev.Created.IsZero() {
|
||||
b.Created = prev.Created
|
||||
}
|
||||
b := s.newBlob(kind, mime, source, id, blobPath, metaPath, int64(len(data)))
|
||||
|
||||
// A blob already on disk costs nothing more, so dedupe is checked before
|
||||
// the budget rather than after it.
|
||||
_, already := os.Stat(blobPath)
|
||||
if already != nil {
|
||||
s.totalMu.Lock()
|
||||
room := s.total+b.Size <= s.maxTotal
|
||||
if room {
|
||||
s.total += b.Size
|
||||
}
|
||||
s.totalMu.Unlock()
|
||||
if !room {
|
||||
return Blob{}, fmt.Errorf("%w: %d stored, %d budget, %d more asked for",
|
||||
ErrStoreFull, s.Total(), s.maxTotal, b.Size)
|
||||
_, statErr := os.Stat(blobPath)
|
||||
fresh := statErr != nil
|
||||
if fresh {
|
||||
if err := s.reserve(b.Size); err != nil {
|
||||
return Blob{}, err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -216,10 +274,8 @@ func (s *Store) Put(kind Kind, mime, source string, data []byte) (Blob, error) {
|
||||
}
|
||||
if err := writeFile(blobPath, data); err != nil {
|
||||
_ = os.Remove(metaPath)
|
||||
if already != nil {
|
||||
s.totalMu.Lock()
|
||||
s.total -= b.Size
|
||||
s.totalMu.Unlock()
|
||||
if fresh {
|
||||
s.release(b.Size)
|
||||
}
|
||||
return Blob{}, err
|
||||
}
|
||||
@@ -264,43 +320,31 @@ func (s *Store) PutFile(kind Kind, mime, source, src string) (Blob, error) {
|
||||
if err != nil {
|
||||
return Blob{}, err
|
||||
}
|
||||
if err := os.MkdirAll(filepath.Dir(blobPath), 0o700); err != nil {
|
||||
if err := os.MkdirAll(filepath.Dir(blobPath), dirPerm); err != nil {
|
||||
return Blob{}, fmt.Errorf("media: create bucket: %w", err)
|
||||
}
|
||||
b := Blob{ID: id, Kind: kind, MIME: mime, Size: info.Size(), Source: source,
|
||||
Created: s.now().UTC(), Path: blobPath}
|
||||
if prev, err := readMeta(metaPath); err == nil && !prev.Created.IsZero() {
|
||||
b.Created = prev.Created
|
||||
}
|
||||
_, already := os.Stat(blobPath)
|
||||
if already != nil {
|
||||
s.totalMu.Lock()
|
||||
room := s.total+b.Size <= s.maxTotal
|
||||
if room {
|
||||
s.total += b.Size
|
||||
}
|
||||
s.totalMu.Unlock()
|
||||
if !room {
|
||||
return Blob{}, fmt.Errorf("%w: %d stored, %d budget, %d more asked for",
|
||||
ErrStoreFull, s.Total(), s.maxTotal, b.Size)
|
||||
b := s.newBlob(kind, mime, source, id, blobPath, metaPath, info.Size())
|
||||
_, statErr := os.Stat(blobPath)
|
||||
fresh := statErr != nil
|
||||
if fresh {
|
||||
if err := s.reserve(b.Size); err != nil {
|
||||
return Blob{}, err
|
||||
}
|
||||
}
|
||||
if err := writeMeta(metaPath, b); err != nil {
|
||||
return Blob{}, err
|
||||
}
|
||||
if already == nil {
|
||||
if !fresh {
|
||||
// Same bytes already here. Drop the spool copy.
|
||||
_ = os.Remove(src)
|
||||
return b, nil
|
||||
}
|
||||
if err := os.Chmod(src, 0o600); err != nil {
|
||||
if err := os.Chmod(src, filePerm); err != nil {
|
||||
return Blob{}, fmt.Errorf("media: chmod spool: %w", err)
|
||||
}
|
||||
if err := os.Rename(src, blobPath); err != nil {
|
||||
_ = os.Remove(metaPath)
|
||||
s.totalMu.Lock()
|
||||
s.total -= b.Size
|
||||
s.totalMu.Unlock()
|
||||
s.release(b.Size)
|
||||
return Blob{}, fmt.Errorf("media: move spool: %w", err)
|
||||
}
|
||||
return b, nil
|
||||
@@ -311,15 +355,15 @@ func (s *Store) PutFile(kind Kind, mime, source, src string) (Blob, error) {
|
||||
// looks at it and List never reports it; PutFile is what turns it into a blob.
|
||||
// The caller owns removing it if it never gets that far.
|
||||
func (s *Store) SpoolFile(prefix string) (*os.File, error) {
|
||||
dir := filepath.Join(s.dir, "spool")
|
||||
if err := os.MkdirAll(dir, 0o700); err != nil {
|
||||
dir := filepath.Join(s.dir, spoolName)
|
||||
if err := os.MkdirAll(dir, dirPerm); err != nil {
|
||||
return nil, fmt.Errorf("media: create spool: %w", err)
|
||||
}
|
||||
f, err := os.CreateTemp(dir, prefix+"-*")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("media: spool: %w", err)
|
||||
}
|
||||
if err := f.Chmod(0o600); err != nil {
|
||||
if err := f.Chmod(filePerm); err != nil {
|
||||
f.Close()
|
||||
return nil, fmt.Errorf("media: chmod spool: %w", err)
|
||||
}
|
||||
@@ -346,9 +390,8 @@ func (s *Store) Get(id string) (Blob, error) {
|
||||
if !validID(id) {
|
||||
return Blob{}, ErrBadID
|
||||
}
|
||||
for _, kind := range []Kind{KindImage, KindAudio} {
|
||||
metaPath := filepath.Join(s.dir, string(kind), id[:2], id+".json")
|
||||
b, err := readMeta(metaPath)
|
||||
for _, kind := range allKinds {
|
||||
b, err := readMeta(filepath.Join(s.bucket(kind, id), id+metaExt))
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
@@ -382,7 +425,7 @@ func (s *Store) Read(id string) (Blob, []byte, error) {
|
||||
// inside the retention window) that is cheap, and it means the sidecars are the
|
||||
// single source of truth with no index to fall out of sync.
|
||||
func (s *Store) List(kind Kind) ([]Blob, error) {
|
||||
kinds := []Kind{KindImage, KindAudio}
|
||||
kinds := allKinds
|
||||
if kind != "" {
|
||||
if !kind.Valid() {
|
||||
return nil, ErrBadKind
|
||||
@@ -391,17 +434,7 @@ func (s *Store) List(kind Kind) ([]Blob, error) {
|
||||
}
|
||||
var out []Blob
|
||||
for _, k := range kinds {
|
||||
root := filepath.Join(s.dir, string(k))
|
||||
err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
if errors.Is(err, fs.ErrNotExist) {
|
||||
return nil // kind never used; not an error
|
||||
}
|
||||
return err
|
||||
}
|
||||
if d.IsDir() || !strings.HasSuffix(path, ".json") {
|
||||
return nil
|
||||
}
|
||||
err := s.walkSidecars(k, func(path string, _ fs.DirEntry) error {
|
||||
b, err := readMeta(path)
|
||||
if err != nil {
|
||||
return nil // a corrupt sidecar is skipped, not fatal
|
||||
@@ -431,8 +464,8 @@ func (s *Store) Delete(id string) error {
|
||||
if !validID(id) {
|
||||
return ErrBadID
|
||||
}
|
||||
for _, kind := range []Kind{KindImage, KindAudio} {
|
||||
bucket := filepath.Join(s.dir, string(kind), id[:2])
|
||||
for _, kind := range allKinds {
|
||||
bucket := s.bucket(kind, id)
|
||||
entries, err := os.ReadDir(bucket)
|
||||
if err != nil {
|
||||
continue
|
||||
@@ -441,21 +474,17 @@ func (s *Store) Delete(id string) error {
|
||||
if !strings.HasPrefix(e.Name(), id) {
|
||||
continue
|
||||
}
|
||||
path := filepath.Join(bucket, e.Name())
|
||||
// Only the bytes count against the budget, so the sidecar's own
|
||||
// size is never given back.
|
||||
var size int64
|
||||
if info, err := e.Info(); err == nil && !strings.HasSuffix(e.Name(), ".json") {
|
||||
if info, err := e.Info(); err == nil && !strings.HasSuffix(e.Name(), metaExt) {
|
||||
size = info.Size()
|
||||
}
|
||||
if err := os.Remove(path); err != nil && !errors.Is(err, fs.ErrNotExist) {
|
||||
if err := os.Remove(filepath.Join(bucket, e.Name())); err != nil && !errors.Is(err, fs.ErrNotExist) {
|
||||
return fmt.Errorf("media: delete %s: %w", shortID(id), err)
|
||||
}
|
||||
if size > 0 {
|
||||
s.totalMu.Lock()
|
||||
s.total -= size
|
||||
if s.total < 0 {
|
||||
s.total = 0
|
||||
}
|
||||
s.totalMu.Unlock()
|
||||
s.release(size)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -498,20 +527,9 @@ func (s *Store) Prune() (int, error) {
|
||||
// Put racing a Prune does not lose its bytes.
|
||||
func (s *Store) pruneOrphans(known map[string]bool, now time.Time) (int, error) {
|
||||
deleted := 0
|
||||
for _, kind := range []Kind{KindImage, KindAudio} {
|
||||
root := filepath.Join(s.dir, string(kind))
|
||||
err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
if errors.Is(err, fs.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
if d.IsDir() || strings.HasSuffix(path, ".json") {
|
||||
return nil
|
||||
}
|
||||
name := d.Name()
|
||||
id, _, _ := strings.Cut(name, ".")
|
||||
for _, kind := range allKinds {
|
||||
err := s.walkBlobFiles(kind, func(path string, d fs.DirEntry) error {
|
||||
id, _, _ := strings.Cut(d.Name(), ".")
|
||||
if known[id] {
|
||||
return nil
|
||||
}
|
||||
@@ -525,12 +543,7 @@ func (s *Store) pruneOrphans(known map[string]bool, now time.Time) (int, error)
|
||||
if err := os.Remove(path); err != nil && !errors.Is(err, fs.ErrNotExist) {
|
||||
return err
|
||||
}
|
||||
s.totalMu.Lock()
|
||||
s.total -= info.Size()
|
||||
if s.total < 0 {
|
||||
s.total = 0
|
||||
}
|
||||
s.totalMu.Unlock()
|
||||
s.release(info.Size())
|
||||
deleted++
|
||||
return nil
|
||||
})
|
||||
@@ -541,13 +554,42 @@ func (s *Store) pruneOrphans(known map[string]bool, now time.Time) (int, error)
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
// walkSidecars visits every sidecar of one kind, and walkBlobFiles every file
|
||||
// that is not one. The sidecars are the store's index and the rest are the
|
||||
// bytes, so a walker always wants exactly one of the two.
|
||||
func (s *Store) walkSidecars(kind Kind, fn func(path string, d fs.DirEntry) error) error {
|
||||
return s.walkKind(kind, true, fn)
|
||||
}
|
||||
|
||||
func (s *Store) walkBlobFiles(kind Kind, fn func(path string, d fs.DirEntry) error) error {
|
||||
return s.walkKind(kind, false, fn)
|
||||
}
|
||||
|
||||
// walkKind walks one kind's directory tree. A kind that was never used has no
|
||||
// directory, which is silence rather than an error.
|
||||
func (s *Store) walkKind(kind Kind, sidecars bool, fn func(path string, d fs.DirEntry) error) error {
|
||||
root := filepath.Join(s.dir, string(kind))
|
||||
return filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
if errors.Is(err, fs.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
if d.IsDir() || strings.HasSuffix(path, metaExt) != sidecars {
|
||||
return nil
|
||||
}
|
||||
return fn(path, d)
|
||||
})
|
||||
}
|
||||
|
||||
// paths returns the blob and sidecar paths for an id.
|
||||
func (s *Store) paths(kind Kind, id, mime string) (blobPath, metaPath string, err error) {
|
||||
if !validID(id) {
|
||||
return "", "", ErrBadID
|
||||
}
|
||||
bucket := filepath.Join(s.dir, string(kind), id[:2])
|
||||
return filepath.Join(bucket, id+extFor(mime, kind)), filepath.Join(bucket, id+".json"), nil
|
||||
bucket := s.bucket(kind, id)
|
||||
return filepath.Join(bucket, id+extFor(mime, kind)), filepath.Join(bucket, id+metaExt), nil
|
||||
}
|
||||
|
||||
// locate finds the stored bytes for an id whose extension we do not know,
|
||||
@@ -556,14 +598,14 @@ func (s *Store) locate(kind Kind, id string) (string, error) {
|
||||
if !validID(id) {
|
||||
return "", ErrBadID
|
||||
}
|
||||
bucket := filepath.Join(s.dir, string(kind), id[:2])
|
||||
bucket := s.bucket(kind, id)
|
||||
entries, err := os.ReadDir(bucket)
|
||||
if err != nil {
|
||||
return "", ErrNotFound
|
||||
}
|
||||
for _, e := range entries {
|
||||
name := e.Name()
|
||||
if strings.HasPrefix(name, id) && !strings.HasSuffix(name, ".json") {
|
||||
if strings.HasPrefix(name, id) && !strings.HasSuffix(name, metaExt) {
|
||||
return filepath.Join(bucket, name), nil
|
||||
}
|
||||
}
|
||||
@@ -573,7 +615,7 @@ func (s *Store) locate(kind Kind, id string) (string, error) {
|
||||
// validID guards every path built from an id. Without it a caller-supplied id
|
||||
// is a path traversal: Get("../../etc/passwd") would read outside the store.
|
||||
func validID(id string) bool {
|
||||
if len(id) != 64 {
|
||||
if len(id) != idLen {
|
||||
return false
|
||||
}
|
||||
for i := 0; i < len(id); i++ {
|
||||
@@ -621,7 +663,7 @@ func writeFile(path string, data []byte) error {
|
||||
return fmt.Errorf("media: temp: %w", err)
|
||||
}
|
||||
defer os.Remove(tmp.Name())
|
||||
if err := tmp.Chmod(0o600); err != nil {
|
||||
if err := tmp.Chmod(filePerm); err != nil {
|
||||
tmp.Close()
|
||||
return fmt.Errorf("media: chmod: %w", err)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user