From 4ba1b7b2bd4d76d72e2e39f8564e4f6bc894cce4 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Tue, 3 Nov 2026 15:54:24 +0000 Subject: [PATCH] feat(addons): move connector file bytes by handle A connector add-on that declares files.transfer no longer carries file bytes in the connector envelope. The host puts a random handle in the request of the op media (write) or send_file (read), and the add-on streams the file with two new host operations, connector.file.put and connector.file.get, in chunks of at most 256 KiB. The process that runs the add-on registers the handle when the call arrives and ends it when the call returns. A handle is bound to the add-on, so another add-on gets the same answer as for an unknown handle. An unfinished inbound file is dropped with its partial bytes, and a handle with no call for two minutes expires. Inbound bytes go to the blob store of the workspace, outbound bytes come from it, so a cluster needs the shared object store and no shared disk. The file is never held in memory. The host overwrites the size and hash of an inbound file with its own record. The change is additive: the field files.transfer needs manifest API 2, a kind without it keeps the inline path and its 1 MiB cap, and the host never sends it a handle. A call that carries a handle may run up to 15 minutes. The ceiling of a file also stays under the largest file of a workspace. Assisted-by: Claude:claude-sonnet-5-5 go test --- addons/internal/connector/route.go | 48 ++ addons/internal/connector/transfer.go | 154 ++++++ addons/internal/connector/transfer_test.go | 163 +++++++ core/application/addon_gateway.go | 2 +- core/application/addon_policy_reader.go | 10 + core/application/application.go | 10 +- core/application/studio_connectors.go | 26 +- core/application/studio_files.go | 1 + core/services/addons/gateway/connectors.go | 86 +++- .../addons/gateway/connectors_test.go | 2 +- .../gateway/connectors_transfer_test.go | 111 +++++ core/services/addons/gateway/gateway.go | 5 + core/services/addons/hostops/admission.go | 2 +- .../services/addons/hostops/admission_test.go | 1 + core/services/addons/hostops/broker.go | 4 + core/services/addons/hostops/transfer.go | 450 ++++++++++++++++++ core/services/addons/hostops/transfer_test.go | 284 +++++++++++ core/services/addons/runtime/client.go | 2 +- core/services/studio/connectors/scope.go | 10 + pkg/addons/contract/connector.go | 20 + pkg/addons/contract/connectorfiles.go | 76 +++ 21 files changed, 1453 insertions(+), 14 deletions(-) create mode 100644 addons/internal/connector/transfer.go create mode 100644 addons/internal/connector/transfer_test.go create mode 100644 core/services/addons/gateway/connectors_transfer_test.go create mode 100644 core/services/addons/hostops/transfer.go create mode 100644 core/services/addons/hostops/transfer_test.go diff --git a/addons/internal/connector/route.go b/addons/internal/connector/route.go index 84db92d7fe..af84a13a30 100644 --- a/addons/internal/connector/route.go +++ b/addons/internal/connector/route.go @@ -11,6 +11,7 @@ import ( "context" "errors" "fmt" + "io" "time" "github.com/mudler/LocalAI/pkg/addons/contract" @@ -111,6 +112,13 @@ type Splitter interface { // means. A request for another kind, without a token or with an unknown op is // refused. func Serve(ctx context.Context, p Platform, req contract.ConnectorRequest) (contract.ConnectorResponse, error) { + return ServeWith(ctx, p, nil, req) +} + +// ServeWith is Serve for an add-on that has the host operation channel, so it +// can move file bytes by handle (see ConnectorTransfer). A request with a +// handle that the Platform cannot serve is refused. +func ServeWith(ctx context.Context, p Platform, host Host, req contract.ConnectorRequest) (contract.ConnectorResponse, error) { if req.Kind != p.Kind() || req.Token == "" { return contract.ConnectorResponse{}, fmt.Errorf("%w: unknown kind or no token", contract.ErrRefused) } @@ -149,6 +157,18 @@ func Serve(ctx context.Context, p Platform, req contract.ConnectorRequest) (cont out.BotID, out.BotUsername = bot.ID, bot.Username } case contract.ConnectorOpSendFile: + if req.Transfer != nil { + ss, ok := p.(StreamSender) + if !ok || req.File != nil || req.Transfer.Handle == "" || req.Transfer.Size <= 0 || req.Transfer.Name != contract.SafeFileName(req.Transfer.Name) { + return contract.ConnectorResponse{}, fmt.Errorf("%w: no file, or one this platform cannot send by handle", contract.ErrRefused) + } + err = Retry(ctx, func() error { + id, err := ss.SendStream(ctx, req, Get(ctx, host, req.Transfer)) + out.MessageID = id + return err + }) + break + } fs, ok := p.(FileSender) if !ok || !validFile(req.File) { return contract.ConnectorResponse{}, fmt.Errorf("%w: no file, or one this platform cannot send", contract.ErrRefused) @@ -159,6 +179,10 @@ func Serve(ctx context.Context, p Platform, req contract.ConnectorRequest) (cont return err }) case contract.ConnectorOpMedia: + if req.Transfer != nil { + out.Received, err = receive(ctx, p, host, req) + break + } m, ok := p.(MediaReader) if !ok || req.Media == "" { return contract.ConnectorResponse{}, fmt.Errorf("%w: no media", contract.ErrRefused) @@ -188,12 +212,36 @@ func Serve(ctx context.Context, p Platform, req contract.ConnectorRequest) (cont default: return contract.ConnectorResponse{}, fmt.Errorf("%w: unknown op", contract.ErrRefused) } + if errors.Is(err, contract.ErrRefused) { + return contract.ConnectorResponse{}, err + } if err != nil { return contract.ConnectorResponse{Error: Typed(err)}, nil } return out, nil } +// receive streams the file of a reference to the host by the handle. +func receive(ctx context.Context, p Platform, host Host, req contract.ConnectorRequest) (*contract.ConnectorReceived, error) { + sr, ok := p.(StreamReader) + if !ok || req.Media == "" || req.Transfer.Handle == "" || req.Transfer.MaxBytes <= 0 { + return nil, fmt.Errorf("%w: no media, or a platform that cannot stream it", contract.ErrRefused) + } + m, err := sr.MediaStream(ctx, req.Token, req.Media, req.Transfer.MaxBytes) + if err != nil { + return nil, err + } + defer func() { _ = m.Body.Close() }() + if m.Size > req.Transfer.MaxBytes { + return nil, ErrTooLarge + } + // One byte past the bound shows a file that is larger than it said. + if _, err := Put(ctx, host, req.Transfer, io.LimitReader(m.Body, req.Transfer.MaxBytes+1)); err != nil { + return nil, err + } + return &contract.ConnectorReceived{Name: contract.SafeFileName(m.Name), MIME: m.MIME}, nil +} + type optionsKey struct{} // OptionsFrom returns the options in effect that the host sent with the op diff --git a/addons/internal/connector/transfer.go b/addons/internal/connector/transfer.go new file mode 100644 index 0000000000..77c6c3becc --- /dev/null +++ b/addons/internal/connector/transfer.go @@ -0,0 +1,154 @@ +// SPDX-License-Identifier: MIT +package connector + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + + "github.com/mudler/LocalAI/pkg/addons/contract" +) + +// Host is the host operation channel of the add-on. *hostclient.Client +// implements it. +type Host interface { + Do(ctx context.Context, operationID, op string, args, out any) error +} + +// transferTries is how often one chunk is sent before the transfer fails. A +// repeat is safe: the host answers where it is, and a chunk it already has is +// not written twice. +const transferTries = 3 + +// ErrNoHost means the add-on got a handle but has no host operation channel. +var ErrNoHost = errors.New("the host gave this add-on no channel for file transfer") + +func hostError(err error) error { + var he *contract.HostError + if errors.As(err, &he) && he.Code == contract.HostTooLarge { + return fmt.Errorf("%w: %v", ErrTooLarge, err) + } + return err +} + +// Put streams r to the host through the handle of an inbound file and ends +// the file. It holds one chunk in memory, whatever the size of the file, and +// returns the bytes sent. The host refuses the file when it passes the bound +// of the handle, and Put then returns ErrTooLarge. +func Put(ctx context.Context, h Host, t *contract.ConnectorTransfer, r io.Reader) (int64, error) { + if h == nil { + return 0, ErrNoHost + } + buf := make([]byte, contract.TransferChunkBytes) + var off int64 + for { + n, rerr := io.ReadFull(r, buf) + last := rerr != nil // a short read or the end + if rerr != nil && !errors.Is(rerr, io.EOF) && !errors.Is(rerr, io.ErrUnexpectedEOF) { + return off, rerr + } + if n == 0 && !last { + continue + } + var res contract.ConnectorFilePutResult + var err error + for try := 0; try < transferTries; try++ { + err = h.Do(ctx, t.Handle, contract.ConnectorFilePutOp, contract.ConnectorFilePut{Handle: t.Handle, Offset: off, Data: buf[:n], Final: last}, &res) + if err == nil || ctx.Err() != nil || errors.Is(hostError(err), ErrTooLarge) { + break + } + } + if err != nil { + return off, hostError(err) + } + if res.Received != off+int64(n) { + // The host is not where the add-on is, and a stream cannot be + // rewound; the platform download starts again. + return off, fmt.Errorf("the host has %d bytes and the add-on sent %d", res.Received, off+int64(n)) + } + off += int64(n) + if last { + return off, nil + } + } +} + +// Get returns a reader of the outbound file of the handle. Each Read pulls at +// most one chunk, so a platform upload that reads from it streams the file +// and holds one chunk. A repeated read of a chunk is safe. +func Get(ctx context.Context, h Host, t *contract.ConnectorTransfer) io.Reader { + return &pull{ctx: ctx, h: h, t: t} +} + +type pull struct { + ctx context.Context + h Host + t *contract.ConnectorTransfer + off int64 + buf []byte + eof bool +} + +func (p *pull) Read(b []byte) (int, error) { + if p.h == nil { + return 0, ErrNoHost + } + for len(p.buf) == 0 { + if p.eof { + return 0, io.EOF + } + var res contract.ConnectorFileGetResult + var err error + for try := 0; try < transferTries; try++ { + err = p.h.Do(p.ctx, p.t.Handle, contract.ConnectorFileGetOp, contract.ConnectorFileGet{Handle: p.t.Handle, Offset: p.off, Length: contract.TransferChunkBytes}, &res) + if err == nil || p.ctx.Err() != nil { + break + } + } + if err != nil { + return 0, err + } + p.buf, p.eof = res.Data, res.EOF + p.off += int64(len(res.Data)) + if len(res.Data) == 0 && !res.EOF { + return 0, errors.New("the host sent an empty chunk before the end of the file") + } + } + n := copy(b, p.buf) + p.buf = p.buf[n:] + return n, nil +} + +// Sniff returns a reader that gives the same bytes as r and the first bytes +// of r, at most n, for a platform that decides the type of a file from them. +func Sniff(r io.Reader, n int) (io.Reader, []byte) { + br := bufio.NewReaderSize(r, max(n, 16)) + head, _ := br.Peek(n) + return br, head +} + +// MediaStream is a file that a platform streams to the add-on. +type MediaStream struct { + Name, MIME string + // Size is the size the platform states, or zero when it states none. + Size int64 + // Body is the stream of bytes; Serve closes it. + Body io.ReadCloser +} + +// StreamReader is a Platform that streams the files of messages for the op +// media when the host gives a handle. max is the bound of the handle: the +// platform should refuse a larger file as soon as it knows the size. A +// Platform that does not implement it keeps the inline path. +type StreamReader interface { + MediaStream(ctx context.Context, token, ref string, max int64) (MediaStream, error) +} + +// StreamSender is a Platform that sends a file for the op send_file from a +// stream. body is the file of req.Transfer: t.Size bytes. Serve builds a new +// body for every attempt. +type StreamSender interface { + SendStream(ctx context.Context, req contract.ConnectorRequest, body io.Reader) (string, error) +} diff --git a/addons/internal/connector/transfer_test.go b/addons/internal/connector/transfer_test.go new file mode 100644 index 0000000000..dc857b1e04 --- /dev/null +++ b/addons/internal/connector/transfer_test.go @@ -0,0 +1,163 @@ +// SPDX-License-Identifier: MIT +package connector_test + +import ( + "bytes" + "context" + "errors" + "io" + "strings" + + "github.com/mudler/LocalAI/addons/internal/connector" + "github.com/mudler/LocalAI/pkg/addons/contract" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// memHost is the host side of a handle in memory: it keeps what was put, and +// serves what is to be got. fail makes the next calls fail. +type memHost struct { + got bytes.Buffer + final bool + src []byte + fail int + max int64 + tooLarge bool + calls int +} + +func (h *memHost) Do(_ context.Context, _, op string, args, out any) error { + h.calls++ + if h.fail > 0 { + h.fail-- + return errors.New("network") + } + switch a := args.(type) { + case contract.ConnectorFilePut: + if h.tooLarge || (h.max > 0 && int64(h.got.Len()+len(a.Data)) > h.max) { + return &contract.HostError{Code: contract.HostTooLarge, Message: "too large"} + } + if a.Offset == int64(h.got.Len()) { + h.got.Write(a.Data) + h.final = a.Final + } + *(out.(*contract.ConnectorFilePutResult)) = contract.ConnectorFilePutResult{Received: int64(h.got.Len())} + case contract.ConnectorFileGet: + end := min(a.Offset+a.Length, int64(len(h.src))) + *(out.(*contract.ConnectorFileGetResult)) = contract.ConnectorFileGetResult{Data: h.src[a.Offset:end], EOF: end == int64(len(h.src))} + } + return nil +} + +var _ = Describe("file transfer by handle", func() { + tr := &contract.ConnectorTransfer{Handle: "h", MaxBytes: 1 << 30} + + It("puts a stream in chunks and ends it, also when its size is a multiple of the chunk", func() { + for _, n := range []int{1, contract.TransferChunkBytes - 1, contract.TransferChunkBytes, 3*contract.TransferChunkBytes + 7} { + h := &memHost{} + data := bytes.Repeat([]byte{'x'}, n) + sent, err := connector.Put(context.Background(), h, tr, bytes.NewReader(data)) + Expect(err).NotTo(HaveOccurred()) + Expect(sent).To(BeEquivalentTo(n)) + Expect(h.got.Bytes()).To(Equal(data)) + Expect(h.final).To(BeTrue()) + } + }) + + It("sends a chunk again after a failed call, once the host says where it is", func() { + h := &memHost{fail: 2} + _, err := connector.Put(context.Background(), h, tr, strings.NewReader("hello")) + Expect(err).NotTo(HaveOccurred()) + Expect(h.got.String()).To(Equal("hello")) + h = &memHost{fail: 3} + _, err = connector.Put(context.Background(), h, tr, strings.NewReader("hello")) + Expect(err).To(HaveOccurred(), "three tries, then the transfer fails") + }) + + It("reports a file over the bound of the handle as too large", func() { + _, err := connector.Put(context.Background(), &memHost{tooLarge: true}, tr, strings.NewReader("hello")) + Expect(err).To(MatchError(connector.ErrTooLarge)) + }) + + It("needs a host", func() { + _, err := connector.Put(context.Background(), nil, tr, strings.NewReader("x")) + Expect(err).To(MatchError(connector.ErrNoHost)) + _, err = io.ReadAll(connector.Get(context.Background(), nil, tr)) + Expect(err).To(MatchError(connector.ErrNoHost)) + }) + + It("gets a file chunk by chunk as a reader", func() { + src := bytes.Repeat([]byte("abc"), contract.TransferChunkBytes) + h := &memHost{src: src} + got, err := io.ReadAll(connector.Get(context.Background(), h, tr)) + Expect(err).NotTo(HaveOccurred()) + Expect(got).To(Equal(src)) + Expect(h.calls).To(Equal(3), "one call per chunk") + }) +}) + +// streaming is a Platform that streams. +type streaming struct { + script + body string + size int64 + upload []byte + gotName string +} + +func (s *streaming) MediaStream(_ context.Context, _, _ string, _ int64) (connector.MediaStream, error) { + return connector.MediaStream{Name: "../a.txt", MIME: "text/plain", Size: s.size, Body: io.NopCloser(strings.NewReader(s.body))}, nil +} + +func (s *streaming) SendStream(_ context.Context, req contract.ConnectorRequest, body io.Reader) (string, error) { + s.upload, _ = io.ReadAll(body) + s.gotName = req.Transfer.Name + return "f1", nil +} + +var _ = Describe("the connector route with a handle", func() { + serve := func(p connector.Platform, h connector.Host, req contract.ConnectorRequest) (contract.ConnectorResponse, error) { + req.Kind, req.Token = "chat", "t" + return connector.ServeWith(context.Background(), p, h, req) + } + + It("streams the file of an op media to the host and answers with the name only", func() { + h := &memHost{} + out, err := serve(&streaming{body: "hello", size: 5}, h, contract.ConnectorRequest{Op: contract.ConnectorOpMedia, Media: "m", Transfer: &contract.ConnectorTransfer{Handle: "h", MaxBytes: 100}}) + Expect(err).NotTo(HaveOccurred()) + Expect(out.Error).To(BeNil()) + Expect(out.File).To(BeNil()) + Expect(out.Received).To(Equal(&contract.ConnectorReceived{Name: "a.txt", MIME: "text/plain"})) + Expect(h.got.String()).To(Equal("hello")) + }) + + It("refuses a file the platform says is over the bound, before any byte moves", func() { + h := &memHost{} + out, _ := serve(&streaming{body: "hello", size: 500}, h, contract.ConnectorRequest{Op: contract.ConnectorOpMedia, Media: "m", Transfer: &contract.ConnectorTransfer{Handle: "h", MaxBytes: 100}}) + Expect(out.Error.Code).To(Equal(contract.ConnectorTooLarge)) + Expect(h.calls).To(BeZero()) + }) + + It("stops a platform that states a small size and sends more", func() { + h := &memHost{max: 100} + out, _ := serve(&streaming{body: strings.Repeat("x", 500), size: 5}, h, contract.ConnectorRequest{Op: contract.ConnectorOpMedia, Media: "m", Transfer: &contract.ConnectorTransfer{Handle: "h", MaxBytes: 100}}) + Expect(out.Error.Code).To(Equal(contract.ConnectorTooLarge)) + }) + + It("sends the file of an op send_file from the host's chunks", func() { + p := &streaming{} + h := &memHost{src: []byte("report body")} + out, err := serve(p, h, contract.ConnectorRequest{Op: contract.ConnectorOpSendFile, Chat: "c", Transfer: &contract.ConnectorTransfer{Handle: "h", Name: "r.txt", MIME: "text/plain", Size: 11}}) + Expect(err).NotTo(HaveOccurred()) + Expect(out.MessageID).To(Equal("f1")) + Expect(string(p.upload)).To(Equal("report body")) + Expect(p.gotName).To(Equal("r.txt")) + }) + + It("refuses a handle on a platform that cannot stream, and a name that is not safe", func() { + _, err := serve(&script{}, &memHost{}, contract.ConnectorRequest{Op: contract.ConnectorOpMedia, Media: "m", Transfer: &contract.ConnectorTransfer{Handle: "h", MaxBytes: 1}}) + Expect(err).To(MatchError(contract.ErrRefused)) + _, err = serve(&streaming{}, &memHost{}, contract.ConnectorRequest{Op: contract.ConnectorOpSendFile, Transfer: &contract.ConnectorTransfer{Handle: "h", Name: "../x", Size: 1}}) + Expect(err).To(MatchError(contract.ErrRefused)) + }) +}) diff --git a/core/application/addon_gateway.go b/core/application/addon_gateway.go index 373449f28d..cf751a6b63 100644 --- a/core/application/addon_gateway.go +++ b/core/application/addon_gateway.go @@ -302,7 +302,7 @@ func (a *Application) startAddonGateway() error { if mode == auth.AddonModeLocalOwner { xlog.Warn(localOwnerWarning) } - g := &gateway.Gateway{Limits: c.Limits, Manager: a.addonManager, ConnectorAuthority: connectorAuthority{a}, Revalidate: auth.AddonRevalidate(a.authDB, mode)} + g := &gateway.Gateway{Transfers: a.transfers(), Limits: c.Limits, Manager: a.addonManager, ConnectorAuthority: connectorAuthority{a}, Revalidate: auth.AddonRevalidate(a.authDB, mode)} if cfg.InstanceID != cfg.OwnerID { g.Forward = carrier a.addonGateway = g diff --git a/core/application/addon_policy_reader.go b/core/application/addon_policy_reader.go index cae940d7fb..614f2c6ef4 100644 --- a/core/application/addon_policy_reader.go +++ b/core/application/addon_policy_reader.go @@ -68,6 +68,16 @@ func (a *Application) studioToolPins(ctx context.Context) (map[string]string, er // the owner policy. Without add-ons there is no owner policy and no connector // to move a file, so the default stands. func (a *Application) studioConnectorFilesCeiling(ctx context.Context) (int64, error) { + n, err := a.policyFilesCeiling(ctx) + if err != nil { + return 0, err + } + // A file that a sender shares becomes a file of the chat, so the ceiling + // never passes what one file of a workspace may be. + return min(n, a.studioUploadLimit()), nil +} + +func (a *Application) policyFilesCeiling(ctx context.Context) (int64, error) { if err := ctx.Err(); err != nil { return 0, err } diff --git a/core/application/application.go b/core/application/application.go index 3f5d35d6bb..e6caa36081 100644 --- a/core/application/application.go +++ b/core/application/application.go @@ -26,6 +26,7 @@ import ( "github.com/mudler/LocalAI/core/services/addons/backendlib" "github.com/mudler/LocalAI/core/services/addons/catalog" "github.com/mudler/LocalAI/core/services/addons/gateway" + "github.com/mudler/LocalAI/core/services/addons/hostops" "github.com/mudler/LocalAI/core/services/addons/hostsettings" "github.com/mudler/LocalAI/core/services/addons/packages" "github.com/mudler/LocalAI/core/services/addons/pins" @@ -113,9 +114,12 @@ type Application struct { backendLibrary *backendlib.Installer addonManager *addons.Manager addonHostOps *addonruntime.HostOps - addonDataRoot string // parent of the private data directory of each add-on; empty when none runs here - addonsProbe addonsProbeState - modelsProbe modelsProbeState + // addonTransfers holds the handles of connector file transfers; see transfers. + addonTransfers *hostops.Transfers + addonTransfersOnce sync.Once + addonDataRoot string // parent of the private data directory of each add-on; empty when none runs here + addonsProbe addonsProbeState + modelsProbe modelsProbeState // libraryStore is the library sources file when no add-on owner runs here. libraryStore *catalog.Sources diff --git a/core/application/studio_connectors.go b/core/application/studio_connectors.go index 91a2b79e0f..9b17872ca7 100644 --- a/core/application/studio_connectors.go +++ b/core/application/studio_connectors.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "time" "github.com/mudler/xlog" @@ -63,7 +64,16 @@ func (t connectorTransport) Call(ctx context.Context, scope studio.ConnectorCall // function as a message the manager pulls itself. A dedicated add-on owner // refuses it instead. func (a *Application) RegisterAddonConnector() error { - if a.addonHostOps == nil || a.studioConnectors == nil { + if a.addonHostOps == nil { + return nil + } + // The two file operations serve every connector add-on that declares + // files.transfer; the handles come with the connector calls. They are + // registered wherever add-ons run, also on a dedicated owner. + if err := hostops.RegisterConnectorTransfer(a.addonHostOps.Broker, a.transfers()); err != nil { + return err + } + if a.studioConnectors == nil { return nil } if a.ownsAddonsOnly() { @@ -130,3 +140,17 @@ func (a *Application) startStudioConnectors(svc *studio.Service) { a.studioConnectors, a.studioConnectorsStop = c, cancel go c.Run(ctx) } + +// transfers is the table of live connector file handles of this process. The +// gateway checks and ends them; the broker serves the bytes. Its blob store is +// the one of the workspace, set when the files start. +func (a *Application) transfers() *hostops.Transfers { + a.addonTransfersOnce.Do(func() { + t := hostops.NewTransfers(nil) + t.Audit = func(o hostops.TransferOutcome) { + xlog.Info("Connector file transfer", "direction", o.Direction, "binding", o.BindingID, "bytes", o.Bytes, "result", o.Result, "duration", o.Duration.Round(time.Millisecond)) + } + a.addonTransfers = t + }) + return a.addonTransfers +} diff --git a/core/application/studio_files.go b/core/application/studio_files.go index 82d2641acd..40241708ce 100644 --- a/core/application/studio_files.go +++ b/core/application/studio_files.go @@ -108,6 +108,7 @@ func (a *Application) initStudioFilesWithS3(newS3 func(context.Context, storage. if err != nil { return stop("workspace blob store", err, "Files are off: their storage could not be opened.") } + a.transfers().UseBlobs(blobs) ws, closer, err := a.openWorkspaceIndex(root, blobs) if err != nil { return stop("workspace index", err, "Files are off: the workspace could not be started.") diff --git a/core/services/addons/gateway/connectors.go b/core/services/addons/gateway/connectors.go index 5f10306953..aa5ec1f32f 100644 --- a/core/services/addons/gateway/connectors.go +++ b/core/services/addons/gateway/connectors.go @@ -13,6 +13,7 @@ import ( "time" "github.com/mudler/LocalAI/core/services/addons" + "github.com/mudler/LocalAI/core/services/addons/hostops" "github.com/mudler/LocalAI/core/services/studio/connectors" "github.com/mudler/LocalAI/pkg/addons/contract" ) @@ -51,7 +52,7 @@ func (g *Gateway) connectorSource() ConnectorSource { func (g *Gateway) ConnectorKinds(ctx context.Context) ([]connectors.Kind, error) { if g.Forward != nil { var out []connectors.Kind - err := g.Forward.connectorRequest(ctx, "GET", "/addons/connectors/kinds", nil, &out) + err := g.Forward.connectorRequest(ctx, "GET", "/addons/connectors/kinds", nil, &out, connectorTimeout) return out, err } return g.localConnectorKinds(ctx) @@ -77,13 +78,17 @@ func (g *Gateway) ConnectorCall(ctx context.Context, s connectors.ConnectorCallS req.Token = "" } var out contract.ConnectorResponse - err := g.Forward.connectorRequest(ctx, "POST", "/addons/connectors/call", connectorCall{s, req}, &out) + err := g.Forward.connectorRequest(ctx, "POST", "/addons/connectors/call", connectorCall{s, req}, &out, req.CallTimeout()) return out, err } return g.localConnectorCall(ctx, s, req) } func (g *Gateway) localConnectorCall(ctx context.Context, s connectors.ConnectorCallScope, req contract.ConnectorRequest) (contract.ConnectorResponse, error) { - ctx, cancel := context.WithTimeout(ctx, connectorTimeout) + timeout := connectorTimeout + if req.Transfer != nil { + timeout = req.CallTimeout() + } + ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() source := g.connectorSource() if source == nil || g.ConnectorAuthority == nil { @@ -109,7 +114,18 @@ func (g *Gateway) localConnectorCall(ctx context.Context, s connectors.Connector } return contract.ConnectorResponse{}, ErrForbidden } + if req.Transfer != nil && (kinds[i].Files == nil || !kinds[i].Files.Transfer) { + // A handle only goes to an add-on that declared that it takes one. + return contract.ConnectorResponse{}, ErrForbidden + } + end, err := g.beginTransfer(s, req) + if err != nil { + return contract.ConnectorResponse{}, err + } out, err := source.Connector(ctx, s.AddonID, req) + if end != nil { + out = finishTransfer(end, out) + } if err != nil { // A definite refusal of the add-on keeps its type; anything else could // not tell, so the caller treats it as an outage, with the cause kept. @@ -120,8 +136,8 @@ func (g *Gateway) localConnectorCall(ctx context.Context, s connectors.Connector } return out, nil } -func (c *Carrier) connectorRequest(ctx context.Context, method, path string, body any, out any) error { - ctx, cancel := context.WithTimeout(ctx, connectorTimeout) +func (c *Carrier) connectorRequest(ctx context.Context, method, path string, body any, out any, timeout time.Duration) error { + ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() var raw []byte if body != nil { @@ -136,7 +152,13 @@ func (c *Carrier) connectorRequest(ctx context.Context, method, path string, bod return ErrUnavailable } req.Header.Set("Content-Type", "application/json") - resp, err := c.client.Do(req) + // A call that moves a file runs longer than the client's own timeout; the + // context carries the bound. + client := c.client + if timeout > connectorTimeout { + client = c.long + } + resp, err := client.Do(req) if err != nil { return ErrUnavailable } @@ -172,6 +194,10 @@ func (g *Gateway) serveConnectors(w http.ResponseWriter, r *http.Request) bool { typedError(w, ErrForbidden) return true } + if call.Request.Transfer != nil { + // The server's write timeout is sized for short calls. + _ = http.NewResponseController(w).SetWriteDeadline(time.Now().Add(contract.ConnectorTransferTimeout + time.Minute)) + } out, err := g.localConnectorCall(r.Context(), call.Scope, call.Request) if err != nil { typedError(w, err) @@ -193,3 +219,51 @@ func decodeConnector(r io.Reader, out any) error { } return decode(bytes.NewReader(raw), out) } + +// beginTransfer registers the handle of a call that carries one, on the +// process that runs the add-on. The frontend made the handle and checked the +// chat's access; this process binds it to the add-on of the call, so no other +// add-on can use it, and to the content or the bound the scope names. +func (g *Gateway) beginTransfer(s connectors.ConnectorCallScope, req contract.ConnectorRequest) (func() hostops.TransferOutcome, error) { + if req.Transfer == nil { + return nil, nil + } + grant := hostops.TransferGrant{Handle: req.Transfer.Handle, AddonID: s.AddonID, BindingID: s.BindingID} + switch req.Op { + case contract.ConnectorOpMedia: + grant.Direction, grant.MaxBytes = hostops.TransferIn, req.Transfer.MaxBytes + case contract.ConnectorOpSendFile: + if s.Content == nil || req.File != nil { + return nil, ErrForbidden + } + grant.Direction, grant.SHA, grant.Size = hostops.TransferOut, s.Content.SHA, s.Content.Size + default: + return nil, ErrForbidden + } + if g.Transfers == nil { + return nil, ErrUnavailable + } + end, err := g.Transfers.Begin(grant) + if err != nil { + if errors.Is(err, hostops.ErrBadRequest) || errors.Is(err, hostops.ErrConflict) { + return nil, ErrForbidden + } + return nil, ErrUnavailable + } + return end, nil +} + +// finishTransfer ends the handle after the add-on answered. For an inbound +// file it writes the size and the hash from the host's own record into the +// answer; an add-on that did not finish the file gets no Received. +func finishTransfer(end func() hostops.TransferOutcome, out contract.ConnectorResponse) contract.ConnectorResponse { + o := end() + if out.Received != nil { + if o.Direction != hostops.TransferIn || o.Result != hostops.TransferDone || o.SHA == "" { + out.Received = nil + } else { + out.Received.Size, out.Received.SHA256 = o.Bytes, o.SHA + } + } + return out +} diff --git a/core/services/addons/gateway/connectors_test.go b/core/services/addons/gateway/connectors_test.go index 7978246865..d9029c1b09 100644 --- a/core/services/addons/gateway/connectors_test.go +++ b/core/services/addons/gateway/connectors_test.go @@ -69,7 +69,7 @@ var _ = Describe("Forwarded connector answers", func() { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(status) })) defer server.Close() c := &Carrier{config: CarrierConfig{OwnerAddress: server.URL}, client: server.Client()} - return c.connectorRequest(context.Background(), "GET", "/addons/connectors/kinds", nil, &[]connectors.Kind{}) + return c.connectorRequest(context.Background(), "GET", "/addons/connectors/kinds", nil, &[]connectors.Kind{}, connectorTimeout) } It("keeps a refusal apart from an outage", func() { diff --git a/core/services/addons/gateway/connectors_transfer_test.go b/core/services/addons/gateway/connectors_transfer_test.go new file mode 100644 index 0000000000..c9683de21f --- /dev/null +++ b/core/services/addons/gateway/connectors_transfer_test.go @@ -0,0 +1,111 @@ +// SPDX-License-Identifier: MIT +package gateway + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "net/http" + "net/http/httptest" + + "github.com/mudler/LocalAI/core/services/addons" + "github.com/mudler/LocalAI/core/services/addons/hostops" + "github.com/mudler/LocalAI/core/services/studio/connectors" + "github.com/mudler/LocalAI/core/services/studio/workspace/blob" + "github.com/mudler/LocalAI/pkg/addons/contract" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// handleSource is an add-on that takes handles: it runs fn during the call. +type handleSource struct { + transfer bool + fn func(contract.ConnectorRequest) contract.ConnectorResponse +} + +func (s handleSource) ConnectorKinds(context.Context) ([]addons.ConnectorKind, error) { + k := contract.ConnectorKind{ID: "slack", Delivery: contract.DeliveryPull} + if s.transfer { + k.Files = &contract.ConnectorFiles{Receive: &contract.FileAbility{MaxBytes: 1 << 30}, Transfer: true} + } + return []addons.ConnectorKind{{AddonID: "slack", ConnectorKind: k}}, nil +} + +func (s handleSource) Connector(_ context.Context, _ string, r contract.ConnectorRequest) (contract.ConnectorResponse, error) { + return s.fn(r), nil +} + +var _ = Describe("Connector calls that carry a file handle", func() { + scope := connectors.ConnectorCallScope{AddonID: "slack", Kind: "slack", BindingID: "b1"} + var ( + t *hostops.Transfers + host http.Handler + ) + BeforeEach(func() { + blobs, err := blob.NewFS(GinkgoT().TempDir()) + Expect(err).NotTo(HaveOccurred()) + t = hostops.NewTransfers(blobs) + b := hostops.New(hostops.NewInFlight(), func(string, string) bool { return true }, contract.DefaultLimits()) + Expect(hostops.RegisterConnectorTransfer(b, t)).To(Succeed()) + host = b.Handler("slack", "k", []string{"connector."}) + }) + put := func(h string, off int64, data []byte, final bool) int { + raw, _ := json.Marshal(contract.ConnectorFilePut{Handle: h, Offset: off, Data: data, Final: final}) + body, _ := json.Marshal(contract.HostRequest{OperationID: "o", Op: contract.ConnectorFilePutOp, Arguments: raw}) + req := httptest.NewRequest(http.MethodPost, "/host", bytes.NewReader(body)) + req.Header.Set(contract.HostCredentialHeader, "k") + rec := httptest.NewRecorder() + host.ServeHTTP(rec, req) + return rec.Code + } + media := func(s handleSource, tr *hostops.Transfers) (contract.ConnectorResponse, error) { + g := &Gateway{Connectors: s, ConnectorAuthority: allowAllConnectors{}, Transfers: tr} + return g.ConnectorCall(context.Background(), scope, contract.ConnectorRequest{Op: contract.ConnectorOpMedia, Kind: "slack", Token: "t", + Transfer: &contract.ConnectorTransfer{Handle: "hh", MaxBytes: 100}}) + } + + It("registers the handle for the call and gives the host's size and hash, not the add-on's", func() { + src := handleSource{transfer: true, fn: func(r contract.ConnectorRequest) contract.ConnectorResponse { + Expect(t.Live()).To(Equal(1)) + Expect(put(r.Transfer.Handle, 0, []byte("hello"), true)).To(Equal(http.StatusOK)) + return contract.ConnectorResponse{Received: &contract.ConnectorReceived{Name: "a.txt", Size: 999, SHA256: "forged"}} + }} + out, err := media(src, t) + Expect(err).NotTo(HaveOccurred()) + sum := sha256.Sum256([]byte("hello")) + Expect(out.Received.SHA256).To(Equal(hex.EncodeToString(sum[:]))) + Expect(out.Received.Size).To(BeEquivalentTo(5)) + Expect(t.Live()).To(BeZero(), "the handle ends with the call") + }) + + It("drops an answer for a file the add-on did not finish", func() { + src := handleSource{transfer: true, fn: func(r contract.ConnectorRequest) contract.ConnectorResponse { + Expect(put(r.Transfer.Handle, 0, []byte("half"), false)).To(Equal(http.StatusOK)) + return contract.ConnectorResponse{Received: &contract.ConnectorReceived{Name: "a.txt", Size: 4, SHA256: "forged"}} + }} + out, err := media(src, t) + Expect(err).NotTo(HaveOccurred()) + Expect(out.Received).To(BeNil()) + Expect(t.Live()).To(BeZero()) + }) + + It("refuses a handle for a kind that did not declare transfer, and without a table", func() { + src := handleSource{fn: func(contract.ConnectorRequest) contract.ConnectorResponse { return contract.ConnectorResponse{} }} + _, err := media(src, t) + Expect(err).To(MatchError(ErrForbidden)) + src.transfer = true + _, err = media(src, nil) + Expect(err).To(MatchError(ErrUnavailable)) + }) + + It("refuses a handle on an op that moves no file, and an outbound handle with no content", func() { + src := handleSource{transfer: true, fn: func(contract.ConnectorRequest) contract.ConnectorResponse { return contract.ConnectorResponse{} }} + g := &Gateway{Connectors: src, ConnectorAuthority: allowAllConnectors{}, Transfers: t} + _, err := g.ConnectorCall(context.Background(), scope, contract.ConnectorRequest{Op: contract.ConnectorOpSend, Kind: "slack", Transfer: &contract.ConnectorTransfer{Handle: "x"}}) + Expect(err).To(MatchError(ErrForbidden)) + _, err = g.ConnectorCall(context.Background(), scope, contract.ConnectorRequest{Op: contract.ConnectorOpSendFile, Kind: "slack", Transfer: &contract.ConnectorTransfer{Handle: "x", Size: 1}}) + Expect(err).To(MatchError(ErrForbidden), "the frontend names the content in the scope") + }) +}) diff --git a/core/services/addons/gateway/gateway.go b/core/services/addons/gateway/gateway.go index d0e1ccf6cc..e5c2000201 100644 --- a/core/services/addons/gateway/gateway.go +++ b/core/services/addons/gateway/gateway.go @@ -11,6 +11,7 @@ import ( "sync" "github.com/mudler/LocalAI/core/services/addons" + "github.com/mudler/LocalAI/core/services/addons/hostops" "github.com/mudler/LocalAI/pkg/addons/contract" ) @@ -25,6 +26,10 @@ var ErrNotSent = fmt.Errorf("the add-on is not running now and nothing was sent: // Gateway deliberately admits lifecycle commands only until resource admission // and proposal confirmation are installed. Installation admin is not data access. type Gateway struct { + // Transfers holds the handles of connector file transfers on the process + // that runs the add-ons. Without it a call that carries a handle is + // unavailable. + Transfers *hostops.Transfers Connectors ConnectorSource ConnectorAuthority ConnectorAuthority // HostSettings is owner-only durable policy; frontends forward, never fall back. diff --git a/core/services/addons/hostops/admission.go b/core/services/addons/hostops/admission.go index e9c42beebe..e0805be710 100644 --- a/core/services/addons/hostops/admission.go +++ b/core/services/addons/hostops/admission.go @@ -54,5 +54,5 @@ func Requests(req *contract.HostOperations, registered, grants []string) Request // must not mistake that startup ordering for unsupported operations. The broker // still requires a registered handler and revalidates every request's authority. func SupportedOperations() []string { - return strings.Fields(`finetune.start finetune.get finetune.list finetune.checkpoints finetune.dataset finetune.export finetune.progress finetune.stop quantize.start quantize.get quantize.list quantize.import quantize.progress quantize.stop workspace.materialize workspace.commit workspace.export workspace.import workspace.changes workspace.admin egress.check egress.audit connector.inbound`) + return strings.Fields(`finetune.start finetune.get finetune.list finetune.checkpoints finetune.dataset finetune.export finetune.progress finetune.stop quantize.start quantize.get quantize.list quantize.import quantize.progress quantize.stop workspace.materialize workspace.commit workspace.export workspace.import workspace.changes workspace.admin egress.check egress.audit connector.inbound connector.file.put connector.file.get`) } diff --git a/core/services/addons/hostops/admission_test.go b/core/services/addons/hostops/admission_test.go index a03360ca3e..ebc71d108d 100644 --- a/core/services/addons/hostops/admission_test.go +++ b/core/services/addons/hostops/admission_test.go @@ -37,6 +37,7 @@ func registeredBuiltinOperations() []string { })).To(Succeed()) Expect(RegisterEgress(b, &fakeEgress{})).To(Succeed()) Expect(RegisterConnector(b, admissionInbox{})).To(Succeed()) + Expect(RegisterConnectorTransfer(b, NewTransfers(nil))).To(Succeed()) // Each registration error is checked above, so a duplicate cannot be hidden // by the broker's map. Ops includes dynamic progress/stop and inbound specs. return b.Ops() diff --git a/core/services/addons/hostops/broker.go b/core/services/addons/hostops/broker.go index 39d9c1f866..ec6433160a 100644 --- a/core/services/addons/hostops/broker.go +++ b/core/services/addons/hostops/broker.go @@ -327,6 +327,10 @@ func (b *Broker) serveInbound(w http.ResponseWriter, r *http.Request, spec Spec, status, code, msg = http.StatusForbidden, contract.HostScope, "not allowed" case errors.Is(err, ErrNotFound): status, code, msg = http.StatusNotFound, contract.HostNotFound, "not found" + case errors.Is(err, ErrTransferTooLarge): + status, code, msg = http.StatusRequestEntityTooLarge, contract.HostTooLarge, "the file is larger than the limit" + case errors.Is(err, ErrConflict): + status, code, msg = http.StatusConflict, contract.HostConflict, "conflict" case errors.Is(err, ErrBadRequest): status, code, msg = http.StatusBadRequest, contract.HostBadRequest, err.Error() } diff --git a/core/services/addons/hostops/transfer.go b/core/services/addons/hostops/transfer.go new file mode 100644 index 0000000000..d7abdf586b --- /dev/null +++ b/core/services/addons/hostops/transfer.go @@ -0,0 +1,450 @@ +// SPDX-License-Identifier: MIT +package hostops + +import ( + "bytes" + "context" + "crypto/subtle" + "encoding/json" + "errors" + "fmt" + "io" + "sync" + "time" + + "github.com/mudler/LocalAI/core/services/studio/workspace" + "github.com/mudler/LocalAI/core/services/studio/workspace/blob" + "github.com/mudler/LocalAI/pkg/addons/contract" +) + +// Direction of a transfer, from the host. +const ( + // TransferIn moves a file from an add-on to the host: the add-on puts. + TransferIn = "in" + // TransferOut moves a file from the host to an add-on: the add-on gets. + TransferOut = "out" +) + +// Results of a transfer, as the audit shows them. +const ( + TransferDone = "done" + TransferTooLarge = "too_large" + TransferExpired = "expired" + TransferRefused = "refused" + TransferIncomplete = "incomplete" +) + +// Default times of a handle. +const ( + // DefaultTransferIdle ends a handle that has seen no call for this long. + DefaultTransferIdle = 2 * time.Minute + // DefaultTransferLife ends a handle this long after it was made, whatever + // it does. A connector call is shorter than this. + DefaultTransferLife = time.Hour +) + +// ErrTransferTooLarge means a put went past the bound of its handle. +var ErrTransferTooLarge = errors.New("the file is larger than the limit of the handle") + +// TransferGrant is what the host that issued a handle records about it. The +// owner of the add-ons registers it when a connector call that carries a +// ConnectorTransfer arrives. +type TransferGrant struct { + Handle string + Direction string + // AddonID is the only add-on that may use the handle. + AddonID string + // BindingID is for the audit. + BindingID string + // MaxBytes bounds an inbound file. + MaxBytes int64 + // SHA and Size name the outbound content, a blob of the workspace store. + SHA string + Size int64 +} + +// TransferOutcome is what a handle ended with. It never holds a name or any +// content. +type TransferOutcome struct { + Direction string + BindingID string + Bytes int64 + Result string + Duration time.Duration + // SHA is the hash of a finished inbound file. + SHA string +} + +// FileScan is the hook after the last byte of an inbound file. An error +// rejects the file. It is a no-op until an owner sets one. +type FileScan func(ctx context.Context, sha string, size int64) error + +// Transfers is the table of live handles of one process, and the bytes behind +// them. Inbound bytes stream into the blob store of the workspace, which in a +// cluster is the shared object store, so no node needs a shared disk. Outbound +// bytes stream out of the same store. A file is never held in memory: a put +// or a get moves one chunk. +type Transfers struct { + // Blobs is the blob store of the workspace. UseBlobs sets it when the + // files start, which can be after the table is made. + Blobs blob.Store + Scan FileScan + // Audit gets the outcome of every handle when it ends. Nil: none. + Audit func(TransferOutcome) + // Idle and Life default to DefaultTransferIdle and DefaultTransferLife. + Idle, Life time.Duration + + mu sync.Mutex + m map[string]*xfer +} + +type xfer struct { + t *Transfers + g TransferGrant + start time.Time + + mu sync.Mutex // one call at a time per handle + received int64 + pw *io.PipeWriter + done chan struct{} // closed when the blob Put of an inbound handle returned + sha string + size int64 + putErr error + final bool + rd io.ReadSeekCloser + cancel context.CancelFunc + idle *time.Timer + life *time.Timer + ended bool + result string +} + +// NewTransfers builds a table over the blob store of the workspace. +func NewTransfers(blobs blob.Store) *Transfers { + return &Transfers{Blobs: blobs, m: map[string]*xfer{}} +} + +// UseBlobs sets the blob store. Handles that begin before it is set are +// refused as unavailable. +func (t *Transfers) UseBlobs(b blob.Store) { + t.mu.Lock() + t.Blobs = b + t.mu.Unlock() +} + +func (t *Transfers) store() blob.Store { + t.mu.Lock() + defer t.mu.Unlock() + return t.Blobs +} + +func (t *Transfers) idle() time.Duration { + if t.Idle > 0 { + return t.Idle + } + return DefaultTransferIdle +} + +func (t *Transfers) life() time.Duration { + if t.Life > 0 { + return t.Life + } + return DefaultTransferLife +} + +// Begin registers a handle. The returned function ends it: it drops a file +// that is not finished and reports the outcome. For a finished inbound file the +// outcome holds the hash and the size. Calling it again returns the same +// outcome. A handle that is already registered is refused, so a handle is +// never used for two calls. +func (t *Transfers) Begin(g TransferGrant) (end func() TransferOutcome, err error) { + if t == nil || t.store() == nil { + return nil, errors.New("file transfer is not available here") + } + if g.Handle == "" || len(g.Handle) > contract.MaxTransferHandle || g.AddonID == "" { + return nil, fmt.Errorf("%w: bad handle", ErrBadRequest) + } + switch g.Direction { + case TransferIn: + if g.MaxBytes <= 0 { + return nil, fmt.Errorf("%w: no bound", ErrBadRequest) + } + case TransferOut: + if g.SHA == "" || g.Size <= 0 { + return nil, fmt.Errorf("%w: no content", ErrBadRequest) + } + default: + return nil, fmt.Errorf("%w: bad direction", ErrBadRequest) + } + x := &xfer{t: t, g: g, start: time.Now()} + t.mu.Lock() + if t.m == nil { + t.m = map[string]*xfer{} + } + if _, dup := t.m[g.Handle]; dup { + t.mu.Unlock() + return nil, fmt.Errorf("%w: handle in use", ErrConflict) + } + t.m[g.Handle] = x + t.mu.Unlock() + x.idle = time.AfterFunc(t.idle(), func() { x.finish(TransferExpired) }) + x.life = time.AfterFunc(t.life(), func() { x.finish(TransferExpired) }) + return func() TransferOutcome { return x.finish("") }, nil +} + +// Live reports how many handles exist; a test and the sweep use it. +func (t *Transfers) Live() int { + t.mu.Lock() + defer t.mu.Unlock() + return len(t.m) +} + +func (t *Transfers) lookup(addonID, handle string) (*xfer, error) { + t.mu.Lock() + x := t.m[handle] + t.mu.Unlock() + // The same answer for a handle that does not exist and one of another + // add-on: a handle cannot be probed. + if x == nil || subtle.ConstantTimeCompare([]byte(x.g.AddonID), []byte(addonID)) != 1 { + return nil, ErrNotFound + } + return x, nil +} + +// finish ends the handle once. why is the result when the handle did not end +// by its own end function. +func (x *xfer) finish(why string) TransferOutcome { + x.idle.Stop() + x.life.Stop() + // A put or a get in progress holds mu; an expiry must not wait for a + // stuck one, so it cancels first. + if x.cancel != nil { + x.cancel() + } + if x.pw != nil { + _ = x.pw.CloseWithError(context.Canceled) + } + x.mu.Lock() + defer x.mu.Unlock() + if x.ended { + return x.outcome() + } + x.ended = true + x.t.mu.Lock() + delete(x.t.m, x.g.Handle) + x.t.mu.Unlock() + if x.rd != nil { + _ = x.rd.Close() + } + if x.done != nil { + <-x.done // the blob Put drops its temp file when its reader fails + } + switch { + case x.result != "": + case why != "": + x.result = why + case x.g.Direction == TransferIn && !x.final: + x.result = TransferIncomplete + default: + x.result = TransferDone + } + o := x.outcome() + if x.t.Audit != nil { + x.t.Audit(o) + } + return o +} + +func (x *xfer) outcome() TransferOutcome { + o := TransferOutcome{Direction: x.g.Direction, BindingID: x.g.BindingID, Bytes: x.received, Result: x.result, Duration: time.Since(x.start)} + if x.g.Direction == TransferOut { + o.Bytes = x.size + } + if x.final && x.result != TransferTooLarge && x.result != TransferRefused { + o.SHA = x.sha + } + return o +} + +func (x *xfer) touch() { x.idle.Reset(x.t.idle()) } + +// put writes the next chunk of an inbound file. +func (x *xfer) put(ctx context.Context, p contract.ConnectorFilePut) (int64, error) { + if x.g.Direction != TransferIn { + return 0, ErrNotFound + } + x.mu.Lock() + defer x.mu.Unlock() + if x.ended || x.final { + return 0, ErrConflict + } + x.touch() + if len(p.Data) > contract.TransferChunkBytes { + return 0, fmt.Errorf("%w: chunk too large", ErrBadRequest) + } + if p.Offset != x.received { + // The add-on resumes where the host is. + return x.received, nil + } + if x.received+int64(len(p.Data)) > x.g.MaxBytes { + x.result = TransferTooLarge + x.abortLocked() + return x.received, ErrTransferTooLarge + } + if x.pw == nil { + pr, pw := io.Pipe() + bctx, cancel := context.WithCancel(context.Background()) + x.pw, x.cancel, x.done = pw, cancel, make(chan struct{}) + go func() { + defer close(x.done) + sha, size, err := x.t.store().Put(bctx, pr, x.g.MaxBytes) + x.sha, x.size, x.putErr = sha, size, err + _ = pr.CloseWithError(errors.New("closed")) + }() + } + if len(p.Data) > 0 { + if _, err := io.Copy(x.pw, bytes.NewReader(p.Data)); err != nil { + return x.received, x.failedLocked(err) + } + x.received += int64(len(p.Data)) + } + if !p.Final { + return x.received, nil + } + if x.received == 0 { + x.result = TransferRefused + x.abortLocked() + return 0, fmt.Errorf("%w: an empty file", ErrBadRequest) + } + _ = x.pw.Close() + <-x.done + switch { + case errors.Is(x.putErr, workspace.ErrTooLarge): + x.result = TransferTooLarge + return x.received, ErrTransferTooLarge + case x.putErr != nil: + return x.received, x.putErr + case x.size != x.received: + return x.received, errors.New("the stored size differs from the bytes received") + } + if x.t.Scan != nil { + if err := x.t.Scan(ctx, x.sha, x.size); err != nil { + x.result = TransferRefused + return x.received, fmt.Errorf("%w: the file was refused", ErrBadRequest) + } + } + x.final = true + return x.received, nil +} + +// failedLocked turns a failed write into the error of the blob store, which +// ended the pipe, so that a bound reached inside the store is too large. +func (x *xfer) failedLocked(err error) error { + <-x.done + if errors.Is(x.putErr, workspace.ErrTooLarge) { + x.result = TransferTooLarge + return ErrTransferTooLarge + } + if x.putErr != nil { + return x.putErr + } + return err +} + +func (x *xfer) abortLocked() { + if x.pw != nil { + _ = x.pw.CloseWithError(context.Canceled) + if x.cancel != nil { + x.cancel() + } + <-x.done + } +} + +// get reads the next chunk of an outbound file. +func (x *xfer) get(ctx context.Context, g contract.ConnectorFileGet) (contract.ConnectorFileGetResult, error) { + if x.g.Direction != TransferOut { + return contract.ConnectorFileGetResult{}, ErrNotFound + } + x.mu.Lock() + defer x.mu.Unlock() + if x.ended { + return contract.ConnectorFileGetResult{}, ErrConflict + } + x.touch() + if g.Offset < 0 || g.Offset > x.g.Size || g.Length < 1 || g.Length > contract.TransferChunkBytes { + return contract.ConnectorFileGetResult{}, fmt.Errorf("%w: offset or length", ErrBadRequest) + } + if x.rd == nil { + r, err := x.t.store().Open(ctx, x.g.SHA) + if err != nil { + return contract.ConnectorFileGetResult{}, err + } + n, err := r.Seek(0, io.SeekEnd) + if err != nil || n != x.g.Size { + _ = r.Close() + return contract.ConnectorFileGetResult{}, errors.New("the file changed since the handle was made") + } + x.rd, x.size = r, n + } + if _, err := x.rd.Seek(g.Offset, io.SeekStart); err != nil { + return contract.ConnectorFileGetResult{}, err + } + buf := make([]byte, min(g.Length, x.g.Size-g.Offset)) + n, err := io.ReadFull(x.rd, buf) + if err != nil && !errors.Is(err, io.ErrUnexpectedEOF) && !errors.Is(err, io.EOF) { + return contract.ConnectorFileGetResult{}, err + } + buf = buf[:n] + return contract.ConnectorFileGetResult{Data: buf, EOF: g.Offset+int64(n) >= x.g.Size}, nil +} + +// RegisterConnectorTransfer registers the two operations with which a +// connector add-on moves file bytes by handle. Both are inbound class: the +// credential of the live incarnation authorizes them, and the handle binds +// them to one connector call. The arguments name no user. +func RegisterConnectorTransfer(b *Broker, t *Transfers) error { + if t == nil { + return errors.New("connector file transfer needs a table") + } + decode := func(raw json.RawMessage, v any) error { + d := json.NewDecoder(bytes.NewReader(raw)) + d.DisallowUnknownFields() + if err := d.Decode(v); err != nil { + return ErrBadRequest + } + return nil + } + if err := b.Register(Spec{Op: contract.ConnectorFilePutOp, Scope: ScopeInbound, Timeout: 2 * time.Minute, Handler: func(ctx context.Context, c Call) (json.RawMessage, error) { + var p contract.ConnectorFilePut + if err := decode(c.Arguments, &p); err != nil { + return nil, err + } + x, err := t.lookup(c.AddonID, p.Handle) + if err != nil { + return nil, err + } + got, err := x.put(ctx, p) + if err != nil { + return nil, err + } + return json.Marshal(contract.ConnectorFilePutResult{Received: got}) + }}); err != nil { + return err + } + return b.Register(Spec{Op: contract.ConnectorFileGetOp, Scope: ScopeInbound, Timeout: 2 * time.Minute, Handler: func(ctx context.Context, c Call) (json.RawMessage, error) { + var g contract.ConnectorFileGet + if err := decode(c.Arguments, &g); err != nil { + return nil, err + } + x, err := t.lookup(c.AddonID, g.Handle) + if err != nil { + return nil, err + } + r, err := x.get(ctx, g) + if err != nil { + return nil, err + } + return json.Marshal(r) + }}) +} diff --git a/core/services/addons/hostops/transfer_test.go b/core/services/addons/hostops/transfer_test.go new file mode 100644 index 0000000000..ed88485956 --- /dev/null +++ b/core/services/addons/hostops/transfer_test.go @@ -0,0 +1,284 @@ +// SPDX-License-Identifier: MIT +package hostops + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "math/rand" + "net/http" + "net/http/httptest" + "path/filepath" + "runtime" + "time" + + "github.com/mudler/LocalAI/core/services/storage" + "github.com/mudler/LocalAI/core/services/studio/workspace/blob" + "github.com/mudler/LocalAI/pkg/addons/contract" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +type xferFixture struct { + t *Transfers + h http.Handler + outs []TransferOutcome + blobs blob.Store + dir string + shared *storage.FilesystemStore +} + +func newXferFixture() *xferFixture { + dir := GinkgoT().TempDir() + shared, err := storage.NewFilesystemStore(filepath.Join(dir, "objects")) + Expect(err).NotTo(HaveOccurred()) + // The owner node has its own cache disk; the object store is the only + // thing the nodes share. + blobs := blob.NewObjectStore(shared, filepath.Join(dir, "owner-cache")) + f := &xferFixture{blobs: blobs, dir: dir, shared: shared} + f.t = NewTransfers(blobs) + f.t.Audit = func(o TransferOutcome) { f.outs = append(f.outs, o) } + b := New(NewInFlight(), always, contract.DefaultLimits()) + Expect(RegisterConnectorTransfer(b, f.t)).To(Succeed()) + f.h = b.Handler("slack", "secret", []string{"connector."}) + return f +} + +// call posts one host operation as the add-on addon. +func (f *xferFixture) call(addon, op string, args any) (int, contract.HostResponse) { + raw, _ := json.Marshal(args) + body, _ := json.Marshal(contract.HostRequest{OperationID: "x", Op: op, Arguments: raw}) + req := httptest.NewRequest(http.MethodPost, "/host", bytes.NewReader(body)) + req.Header.Set(contract.HostCredentialHeader, "secret") + rec := httptest.NewRecorder() + h := f.h + if addon != "slack" { + b := New(NewInFlight(), always, contract.DefaultLimits()) + Expect(RegisterConnectorTransfer(b, f.t)).To(Succeed()) + h = b.Handler(addon, "secret", []string{"connector."}) + } + h.ServeHTTP(rec, req) + var out contract.HostResponse + _ = json.Unmarshal(rec.Body.Bytes(), &out) + return rec.Code, out +} + +func (f *xferFixture) put(handle string, off int64, data []byte, final bool) (int, int64) { + code, r := f.call("slack", contract.ConnectorFilePutOp, contract.ConnectorFilePut{Handle: handle, Offset: off, Data: data, Final: final}) + var res contract.ConnectorFilePutResult + if r.Result != nil { + _ = json.Unmarshal(r.Result, &res) + } + return code, res.Received +} + +func (f *xferFixture) get(handle string, off, n int64) (int, contract.ConnectorFileGetResult) { + code, r := f.call("slack", contract.ConnectorFileGetOp, contract.ConnectorFileGet{Handle: handle, Offset: off, Length: n}) + var res contract.ConnectorFileGetResult + if r.Result != nil { + _ = json.Unmarshal(r.Result, &res) + } + return code, res +} + +// chunks streams n bytes of a seeded sequence, chunk bytes at a time, and +// never holds more than one chunk. +func chunks(n, chunk int64, each func(off int64, b []byte, last bool)) string { + rng := rand.New(rand.NewSource(7)) + h := sha256.New() + buf := make([]byte, chunk) + for off := int64(0); off < n; off += chunk { + b := buf[:min(chunk, n-off)] + _, _ = rng.Read(b) + h.Write(b) + each(off, b, off+int64(len(b)) >= n) + } + return hex.EncodeToString(h.Sum(nil)) +} + +func heap() uint64 { + var m runtime.MemStats + runtime.GC() + runtime.ReadMemStats(&m) + return m.HeapAlloc +} + +var _ = Describe("connector file transfer by handle", func() { + const mb = 1 << 20 + + It("moves 20 MB in and 20 MB out with bounded memory", func() { + f := newXferFixture() + end, err := f.t.Begin(TransferGrant{Handle: "h1", Direction: TransferIn, AddonID: "slack", BindingID: "b1", MaxBytes: 50 * mb}) + Expect(err).NotTo(HaveOccurred()) + base, peak := heap(), uint64(0) + var count int + want := chunks(20*mb, contract.TransferChunkBytes, func(off int64, b []byte, last bool) { + code, got := f.put("h1", off, append([]byte(nil), b...), last) + Expect(code).To(Equal(http.StatusOK)) + Expect(got).To(Equal(off + int64(len(b)))) + if count++; count%16 == 0 { + peak = max(peak, heap()) + } + }) + out := end() + Expect(out.Result).To(Equal(TransferDone)) + Expect(out.Bytes).To(Equal(int64(20 * mb))) + Expect(out.SHA).To(Equal(want)) + Expect(peak).To(BeNumerically("<", base+16*mb), "the file is never held in memory") + Expect(f.t.Live()).To(BeZero()) + + // Another node: same object store, its own cache disk. + node2 := blob.NewObjectStore(f.shared, filepath.Join(f.dir, "frontend-cache")) + r, err := node2.Open(GinkgoT().Context(), out.SHA) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = r.Close() }) + h := sha256.New() + n, err := io.Copy(h, r) + Expect(err).NotTo(HaveOccurred()) + Expect(n).To(Equal(int64(20 * mb))) + Expect(hex.EncodeToString(h.Sum(nil))).To(Equal(want)) + + // Out: the add-on pulls the same file by a new handle. + end2, err := f.t.Begin(TransferGrant{Handle: "h2", Direction: TransferOut, AddonID: "slack", SHA: out.SHA, Size: n}) + Expect(err).NotTo(HaveOccurred()) + hh := sha256.New() + var off int64 + base, peak = heap(), 0 + for i := 0; ; i++ { + code, res := f.get("h2", off, contract.TransferChunkBytes) + Expect(code).To(Equal(http.StatusOK)) + hh.Write(res.Data) + off += int64(len(res.Data)) + if i%16 == 0 { + peak = max(peak, heap()) + } + if res.EOF { + break + } + } + Expect(off).To(Equal(n)) + Expect(hex.EncodeToString(hh.Sum(nil))).To(Equal(want)) + Expect(peak).To(BeNumerically("<", base+16*mb)) + Expect(end2().Result).To(Equal(TransferDone)) + Expect(f.outs).To(HaveLen(2)) + }) + + It("resumes at the offset the host has after a gap", func() { + f := newXferFixture() + end, _ := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: mb}) + code, got := f.put("h", 0, []byte("abcd"), false) + Expect(code).To(Equal(http.StatusOK)) + Expect(got).To(BeEquivalentTo(4)) + _, got = f.put("h", 10, []byte("zz"), false) + Expect(got).To(BeEquivalentTo(4), "a gap answers where the host is and writes nothing") + _, got = f.put("h", 4, []byte("efgh"), true) + Expect(got).To(BeEquivalentTo(8)) + o := end() + sum := sha256.Sum256([]byte("abcdefgh")) + Expect(o.SHA).To(Equal(hex.EncodeToString(sum[:]))) + }) + + It("refuses an oversize file and drops the partial one", func() { + f := newXferFixture() + end, _ := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 10}) + code, _ := f.put("h", 0, []byte("12345678"), false) + Expect(code).To(Equal(http.StatusOK)) + code, _ = f.put("h", 8, []byte("9012"), true) + Expect(code).To(Equal(http.StatusRequestEntityTooLarge)) + o := end() + Expect(o.Result).To(Equal(TransferTooLarge)) + Expect(o.SHA).To(BeEmpty()) + code, _ = f.put("h", 0, []byte("x"), false) + Expect(code).To(Equal(http.StatusNotFound), "the handle ended") + }) + + It("refuses a file whose last chunk says it is short of the bytes sent, an empty file, and a write after final", func() { + f := newXferFixture() + end, _ := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + code, _ := f.put("h", 0, nil, true) + Expect(code).To(Equal(http.StatusBadRequest), "empty") + Expect(end().Result).To(Equal(TransferRefused)) + + end, _ = f.t.Begin(TransferGrant{Handle: "g", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + code, _ = f.put("g", 0, []byte("ab"), true) + Expect(code).To(Equal(http.StatusOK)) + code, _ = f.put("g", 2, []byte("cd"), false) + Expect(code).To(Equal(http.StatusConflict), "write after final") + Expect(end().Result).To(Equal(TransferDone)) + }) + + It("refuses another add-on, an unknown handle and the wrong direction with one answer", func() { + f := newXferFixture() + end, _ := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + defer end() + other, _ := f.call("telegram", contract.ConnectorFilePutOp, contract.ConnectorFilePut{Handle: "h", Data: []byte("x")}) + unknown, _ := f.call("slack", contract.ConnectorFilePutOp, contract.ConnectorFilePut{Handle: "nope", Data: []byte("x")}) + Expect(other).To(Equal(http.StatusNotFound)) + Expect(unknown).To(Equal(other)) + code, _ := f.get("h", 0, 10) + Expect(code).To(Equal(http.StatusNotFound), "a write handle cannot read") + }) + + It("does not let a handle be used for two calls", func() { + f := newXferFixture() + end, err := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + Expect(err).NotTo(HaveOccurred()) + _, err = f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + Expect(err).To(HaveOccurred()) + end() + end() // idempotent + Expect(f.outs).To(HaveLen(1)) + _, err = f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + Expect(err).NotTo(HaveOccurred(), "a handle of an ended call is free; a new call makes its own random one") + }) + + It("expires a handle that sees no call and removes the partial file", func() { + f := newXferFixture() + f.t.Idle = 80 * time.Millisecond + _, err := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + Expect(err).NotTo(HaveOccurred()) + code, _ := f.put("h", 0, []byte("partial"), false) + Expect(code).To(Equal(http.StatusOK)) + Eventually(f.t.Live, "2s", "20ms").Should(BeZero()) + code, _ = f.put("h", 7, []byte("x"), false) + Expect(code).To(Equal(http.StatusNotFound)) + Expect(f.outs).To(HaveLen(1)) + Expect(f.outs[0].Result).To(Equal(TransferExpired)) + tmp, _ := filepath.Glob(filepath.Join(f.dir, "owner-cache", "tmp", "*")) + Expect(tmp).To(BeEmpty(), "no partial file is left") + }) + + It("refuses a file the scan hook rejects", func() { + f := newXferFixture() + f.t.Scan = func(_ context.Context, _ string, _ int64) error { return io.ErrClosedPipe } + end, _ := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + code, _ := f.put("h", 0, []byte("virus"), true) + Expect(code).To(Equal(http.StatusBadRequest)) + o := end() + Expect(o.Result).To(Equal(TransferRefused)) + Expect(o.SHA).To(BeEmpty()) + }) + + It("refuses an outbound handle whose content changed size and an offset past the end", func() { + f := newXferFixture() + sha, size, err := f.blobs.Put(GinkgoT().Context(), bytes.NewReader([]byte("hello world")), 0) + Expect(err).NotTo(HaveOccurred()) + bad, _ := f.t.Begin(TransferGrant{Handle: "bad", Direction: TransferOut, AddonID: "slack", SHA: sha, Size: size + 1}) + code, _ := f.get("bad", 0, 5) + Expect(code).To(Equal(http.StatusServiceUnavailable)) + bad() + end, _ := f.t.Begin(TransferGrant{Handle: "ok", Direction: TransferOut, AddonID: "slack", SHA: sha, Size: size}) + defer end() + code, res := f.get("ok", 6, contract.TransferChunkBytes+1) + Expect(code).To(Equal(http.StatusBadRequest), "a length above the chunk is refused") + code, res = f.get("ok", 6, 5) + Expect(code).To(Equal(http.StatusOK)) + Expect(string(res.Data)).To(Equal("world")) + Expect(res.EOF).To(BeTrue()) + code, _ = f.get("ok", size+1, 1) + Expect(code).To(Equal(http.StatusBadRequest)) + }) +}) diff --git a/core/services/addons/runtime/client.go b/core/services/addons/runtime/client.go index fedc6643dc..4c84a49e9c 100644 --- a/core/services/addons/runtime/client.go +++ b/core/services/addons/runtime/client.go @@ -266,7 +266,7 @@ func (c *Client) Connector(ctx context.Context, v contract.ConnectorRequest) (co if c.registration.Capabilities.Connector == nil { return r, ErrNotSent } - e := c.call(ctx, "/connector", nil, v, &r, contract.ConnectorTimeout) + e := c.call(ctx, "/connector", nil, v, &r, v.CallTimeout()) return r, e } func (c *Client) Preview(ctx context.Context, v contract.Envelope) (contract.Preview, error) { diff --git a/core/services/studio/connectors/scope.go b/core/services/studio/connectors/scope.go index 4fc656d0f4..a804b5ef35 100644 --- a/core/services/studio/connectors/scope.go +++ b/core/services/studio/connectors/scope.go @@ -12,6 +12,16 @@ type ConnectorCallScope struct { AgentID string `json:"agentId,omitempty"` AddonID string `json:"addonId"` Kind string `json:"kind"` + // Content names the file an outbound transfer reads: a blob of the + // workspace store. The frontend that checked the chat's access sets it; + // it never reaches the add-on, which gets only the handle. + Content *TransferContent `json:"content,omitempty"` +} + +// TransferContent is the stored content behind an outbound handle. +type TransferContent struct { + SHA string `json:"sha"` + Size int64 `json:"size"` } const ( diff --git a/pkg/addons/contract/connector.go b/pkg/addons/contract/connector.go index d559132fdd..fba032a3da 100644 --- a/pkg/addons/contract/connector.go +++ b/pkg/addons/contract/connector.go @@ -435,6 +435,11 @@ type ConnectorRequest struct { // send_file: the file. Its name is SafeFileName of itself and its data // is 1 to MaxConnectorFileBytes long. File *ConnectorFile `json:"file,omitempty"` + // Transfer is the handle for the bytes of a file, set only for a kind + // that declares files.transfer. media: the add-on writes the file with + // ConnectorFilePutOp. send_file: File is nil and the add-on reads the + // file with ConnectorFileGetOp. + Transfer *ConnectorTransfer `json:"transfer,omitempty"` } // ConnectorFile is a file that the op media returns or the op send_file @@ -468,6 +473,9 @@ type ConnectorResponse struct { BotUsername string `json:"botUsername,omitempty"` // media File *ConnectorFile `json:"file,omitempty"` + // Received answers the op media of a request with a Transfer: the file + // was written by handle. The host adds Size and SHA256 from its record. + Received *ConnectorReceived `json:"received,omitempty"` // history: the messages, oldest first, at most the Limit asked for. // Truncated says that older messages exist that are not returned. Messages []ConnectorHistoryMessage `json:"messages,omitempty"` @@ -503,6 +511,18 @@ type ConnectorRuntime interface { // MaxPollSeconds for the platform. const ConnectorTimeout = (MaxPollSeconds + 20) * time.Second +// ConnectorTransferTimeout is how long a call that carries a Transfer may +// run: it moves a whole file chunk by chunk, and a large file takes minutes. +const ConnectorTransferTimeout = 15 * time.Minute + +// CallTimeout is the time one connector call may run. +func (r ConnectorRequest) CallTimeout() time.Duration { + if r.Transfer != nil { + return ConnectorTransferTimeout + } + return ConnectorTimeout +} + // SafeFileName returns a file name that every platform takes: the last // element of a path, without control characters, quotes or path separators, // cut to MaxConnectorFileName bytes on a character boundary and keeping the diff --git a/pkg/addons/contract/connectorfiles.go b/pkg/addons/contract/connectorfiles.go index bcdc7859f8..dede757f01 100644 --- a/pkg/addons/contract/connectorfiles.go +++ b/pkg/addons/contract/connectorfiles.go @@ -40,6 +40,11 @@ type ConnectorFiles struct { Receive *FileAbility `json:"receive,omitempty"` // Send is the ability to upload files from the bot. Send *FileAbility `json:"send,omitempty"` + // Transfer says that the add-on moves file bytes by a short-lived handle + // (ConnectorTransfer, ConnectorFilePutOp, ConnectorFileGetOp) and not in + // the connector envelope. Only then may a limit pass MaxConnectorFileBytes. + // A kind that declares files without it keeps the inline path. + Transfer bool `json:"transfer,omitempty"` } var mediaTypePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9!#$&^_.+-]{0,63}/(\*|[a-z0-9][a-z0-9!#$&^_.+-]{0,63})$`) @@ -79,3 +84,74 @@ func (f *ConnectorFiles) validate(api VersionRange) error { } return f.Send.validate("send") } + +// Host operations with which an add-on moves the bytes of a file by handle. A +// handle is issued by the host with one connector call (ConnectorRequest.Transfer), +// is single use, bound to that call, that add-on and that binding, and ends +// when the call ends. The add-on needs the operation prefix "connector." in +// its host policy. +const ( + // ConnectorFilePutOp writes the next chunk of an inbound file. + ConnectorFilePutOp = "connector.file.put" + // ConnectorFileGetOp reads the next chunk of an outbound file. + ConnectorFileGetOp = "connector.file.get" + // TransferChunkBytes is the largest chunk of data of one put or get. It + // is far below the payload limit of a host operation after base64. + TransferChunkBytes = 256 << 10 + // MaxTransferHandle bounds the length of a handle. + MaxTransferHandle = 128 +) + +// ConnectorTransfer is the handle that the host gives to the add-on with the +// ops media (inbound, write only) and send_file (outbound, read only). +type ConnectorTransfer struct { + // Handle is random and means nothing to the add-on. + Handle string `json:"handle"` + // MaxBytes is the largest inbound file the host takes, the effective + // limit. The add-on should refuse a larger file before it downloads it. + MaxBytes int64 `json:"maxBytes,omitempty"` + // Name, MIME and Size describe the outbound file. Name is already safe. + Name string `json:"name,omitempty"` + MIME string `json:"mime,omitempty"` + Size int64 `json:"size,omitempty"` +} + +// ConnectorFilePut is the argument object of ConnectorFilePutOp. Offset must +// equal the bytes the host already has; when it does not, the host answers +// with Received and the add-on resumes there. Final ends the file. +type ConnectorFilePut struct { + Handle string `json:"handle"` + Offset int64 `json:"offset"` + Data []byte `json:"data,omitempty"` + Final bool `json:"final,omitempty"` +} + +// ConnectorFilePutResult is the answer of ConnectorFilePutOp. +type ConnectorFilePutResult struct { + Received int64 `json:"received"` +} + +// ConnectorFileGet is the argument object of ConnectorFileGetOp. Length is at +// most TransferChunkBytes. +type ConnectorFileGet struct { + Handle string `json:"handle"` + Offset int64 `json:"offset"` + Length int64 `json:"length"` +} + +// ConnectorFileGetResult is the answer of ConnectorFileGetOp. EOF is true +// when Data ends the file. +type ConnectorFileGetResult struct { + Data []byte `json:"data,omitempty"` + EOF bool `json:"eof,omitempty"` +} + +// ConnectorReceived is what the add-on answers to the op media when it wrote +// the file by handle: Name and MIME. The host that holds the handle overwrites +// Size and SHA256 with its own record, so the add-on cannot choose them. +type ConnectorReceived struct { + Name string `json:"name"` + MIME string `json:"mime,omitempty"` + Size int64 `json:"size,omitempty"` + SHA256 string `json:"sha256,omitempty"` +}