diff --git a/core/services/addons/runtime/client.go b/core/services/addons/runtime/client.go index 4c84a49e9c..406a11d37a 100644 --- a/core/services/addons/runtime/client.go +++ b/core/services/addons/runtime/client.go @@ -24,9 +24,13 @@ import ( // Client is an incarnation-bound capability. Only the supervisor creates one. // Callers must obtain Authority from host authentication, never model arguments. type Client struct { - mu sync.Mutex - accept bool - active int + mu sync.Mutex + accept bool + active int + // polls counts the connector long polls in flight. They have a pool of + // their own (maxPolls) so that a binding that waits up to 25 s does not + // take a slot from the short calls (see admitIn). + polls int observationLeases int observations map[string]context.CancelFunc endpoint, credential string @@ -64,7 +68,7 @@ func (c *Client) drain(ctx context.Context) { defer ticker.Stop() for { c.mu.Lock() - n := c.active + c.observationLeases + n := c.active + c.polls + c.observationLeases c.mu.Unlock() if n == 0 { return @@ -76,10 +80,33 @@ func (c *Client) drain(ctx context.Context) { } } } -func (c *Client) admit() (func(), error) { + +// maxPolls bounds the connector long polls in flight to one add-on. The host +// keeps one poll per binding, so this is also the most pull bindings of one +// add-on. A poll holds its slot for up to contract.MaxPollSeconds, which is +// why it must not draw on limits.Concurrency, the budget of the short calls. +const maxPolls = 64 + +func (c *Client) admit() (func(), error) { return c.admitIn(false) } + +// admitIn admits one call into the pool of short calls or, for a long poll, +// into the poll pool. Both pools stop at withdraw, and the returned function +// frees the slot once. +func (c *Client) admitIn(poll bool) (func(), error) { c.mu.Lock() defer c.mu.Unlock() - if !c.accept || int64(c.active) >= c.limits.Concurrency { + if !c.accept { + return nil, ErrNotSent + } + if poll { + if c.polls >= maxPolls { + return nil, ErrNotSent + } + c.polls++ + var once sync.Once + return func() { once.Do(func() { c.mu.Lock(); c.polls--; c.mu.Unlock() }) }, nil + } + if int64(c.active) >= c.limits.Concurrency { return nil, ErrNotSent } c.active++ @@ -207,7 +234,10 @@ func (c *Client) begin(ctx context.Context, env *contract.Envelope) func() { return c.inflight.BeginRuntime(*env, hostops.Binding{}, false, c.registration.Incarnation) } func (c *Client) call(ctx context.Context, path string, env *contract.Envelope, in, out any, timeout time.Duration) error { - done, e := c.admit() + return c.callIn(false, ctx, path, env, in, out, timeout) +} +func (c *Client) callIn(poll bool, ctx context.Context, path string, env *contract.Envelope, in, out any, timeout time.Duration) error { + done, e := c.admitIn(poll) if e != nil { return e } @@ -266,7 +296,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, v.CallTimeout()) + e := c.callIn(v.Op == contract.ConnectorOpPoll, 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/addons/runtime/connector_test.go b/core/services/addons/runtime/connector_test.go index 1525ff2495..e560570d47 100644 --- a/core/services/addons/runtime/connector_test.go +++ b/core/services/addons/runtime/connector_test.go @@ -51,4 +51,107 @@ var _ = Describe("the connector call of the runtime client", func() { _, err = c.Connector(context.Background(), contract.ConnectorRequest{Op: "me"}) Expect(err).To(MatchError(ErrNotSent)) }) + + Describe("long polls and short calls", func() { + // An add-on whose poll waits until released and whose other ops answer at once. + var release chan struct{} + var started chan struct{} + newPollClient := func() *Client { + release, started = make(chan struct{}), make(chan struct{}, 4*maxPolls) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req contract.ConnectorRequest + Expect(json.NewDecoder(r.Body).Decode(&req)).To(Succeed()) + if req.Op == contract.ConnectorOpPoll { + started <- struct{}{} + select { + case <-release: + case <-r.Context().Done(): + return + } + } + _ = json.NewEncoder(w).Encode(contract.ConnectorResponse{}) + })) + DeferCleanup(srv.Close) + reg := Registration{AddonID: "telegram", Generation: 1, Incarnation: "inc", Protocol: 1, Endpoint: srv.URL, + Capabilities: contract.Capabilities{Connector: &contract.ConnectorCapability{Kinds: []contract.ConnectorKind{{ID: "telegram", Title: "Telegram", Delivery: contract.DeliveryPull}}}}} + c, err := newClient(reg, credential, contract.DefaultLimits()) + Expect(err).NotTo(HaveOccurred()) + c.accept = true + return c + } + poll := func(ctx context.Context, c *Client) chan error { + out := make(chan error, 1) + go func() { + _, err := c.Connector(ctx, contract.ConnectorRequest{Op: contract.ConnectorOpPoll, Kind: "telegram", Token: "t"}) + out <- err + }() + return out + } + short := func(c *Client) error { + _, err := c.Connector(context.Background(), contract.ConnectorRequest{Op: contract.ConnectorOpMe, Kind: "telegram", Token: "t"}) + return err + } + + It("keeps the short-call slots free while more polls than the concurrency limit are open", func() { + c := newPollClient() + Expect(contract.DefaultLimits().Concurrency).To(BeEquivalentTo(4)) + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + var done []chan error + for i := 0; i < 8; i++ { + done = append(done, poll(ctx, c)) + } + for i := 0; i < 8; i++ { + Eventually(started).Should(Receive()) + } + for i := 0; i < 4; i++ { // the full short-call budget is still there + Expect(short(c)).To(Succeed()) + } + close(release) + for _, d := range done { + Eventually(d).Should(Receive(BeNil())) + } + }) + + It("refuses a poll past the poll pool, and frees the slot when a poll is cancelled", func() { + c := newPollClient() + ctx, cancel := context.WithCancel(context.Background()) + DeferCleanup(cancel) + var done []chan error + for i := 0; i < maxPolls; i++ { + done = append(done, poll(ctx, c)) + } + for i := 0; i < maxPolls; i++ { + Eventually(started).Should(Receive()) + } + _, err := c.Connector(context.Background(), contract.ConnectorRequest{Op: contract.ConnectorOpPoll, Kind: "telegram", Token: "t"}) + Expect(err).To(MatchError(ErrNotSent)) + Expect(short(c)).To(Succeed()) + + cancel() + for _, d := range done { + Eventually(d).Should(Receive(Not(BeNil()))) + } + Eventually(func() int { c.mu.Lock(); defer c.mu.Unlock(); return c.polls }).Should(BeZero()) + close(release) + Eventually(poll(context.Background(), c)).Should(Receive(BeNil())) + }) + + It("refuses a poll after the client is withdrawn and waits for open polls to drain", func() { + c := newPollClient() + d := poll(context.Background(), c) + Eventually(started).Should(Receive()) + c.withdraw() + _, err := c.Connector(context.Background(), contract.ConnectorRequest{Op: contract.ConnectorOpPoll, Kind: "telegram", Token: "t"}) + Expect(err).To(MatchError(ErrNotSent)) + Expect(short(c)).To(MatchError(ErrNotSent)) + + drained := make(chan struct{}) + go func() { c.drain(context.Background()); close(drained) }() + Consistently(drained, "50ms").ShouldNot(BeClosed()) + close(release) + Eventually(d).Should(Receive(BeNil())) + Eventually(drained).Should(BeClosed()) + }) + }) })