diff --git a/e2e/tests/prod-trust-boundary.test.ts b/e2e/tests/prod-trust-boundary.test.ts index 0dff062e..0aa13d1a 100644 --- a/e2e/tests/prod-trust-boundary.test.ts +++ b/e2e/tests/prod-trust-boundary.test.ts @@ -104,6 +104,11 @@ const AUTH_REALTIME_INTERNAL_ENDPOINTS = [ method: "POST", path: "/auth-realtime/api/internal/documents/doc-e2e/branches/branch-e2e/operations", }, + { + name: "a branch flush", + method: "POST", + path: "/auth-realtime/api/internal/documents/doc-e2e/branches/branch-e2e/flush", + }, ] as const // spellings of a blocked path that a proxy is expected to still recognise. diff --git a/server/auth-realtime/src/routes.test.ts b/server/auth-realtime/src/routes.test.ts index f449f975..58b1ee3e 100644 --- a/server/auth-realtime/src/routes.test.ts +++ b/server/auth-realtime/src/routes.test.ts @@ -1457,6 +1457,40 @@ describe("createRoutes", () => { }) }) + describe("POST /internal/documents/:documentId/branches/:branchId/flush", () => { + it("flushes the branch and answers once core has it", async ({ + expect, + }) => { + const { app, flushDocument } = build() + + const res = await app.request( + "/internal/documents/doc1/branches/b1/flush", + { method: "POST" }, + ) + + expect(res.status).toBe(204) + expect( + flushDocument.mock.calls.map((call) => call[0]), + ).toEqual(["doc1-b1"]) + }) + + it("reports a flush that failed", async ({ expect }) => { + const flushDocument = vi + .fn() + .mockRejectedValue( + new Error("core unreachable"), + ) + const { app } = build({ flushDocument }) + + const res = await app.request( + "/internal/documents/doc1/branches/b1/flush", + { method: "POST" }, + ) + + expect(res.status).toBe(500) + }) + }) + describe("POST /internal/documents/:documentId/branches/:branchId/operations", () => { const path = "/internal/documents/doc1/branches/b1/operations" diff --git a/server/auth-realtime/src/routes.ts b/server/auth-realtime/src/routes.ts index 97c742be..5e9caa16 100644 --- a/server/auth-realtime/src/routes.ts +++ b/server/auth-realtime/src/routes.ts @@ -658,6 +658,25 @@ export function createRoutes({ }) }) + // core calls this before it reads a branch it is about to change, so + // it reads what the editors hold. + app.post( + "/internal/documents/:documentId/branches/:branchId/flush", + async (c) => { + const branch = await flushed(c, () => + flushBranch( + c.req.param("documentId"), + c.req.param("branchId"), + ), + ) + if (branch instanceof Response) { + return branch + } + + return c.body(null, 204) + }, + ) + // Server-to-server endpoint called by the Go assistant to apply // edit operations to a live Y.Doc. Each request opens (or reuses) a // direct connection to the document, runs the batch inside a single diff --git a/server/core/internal/assistant/edit/client.go b/server/core/internal/assistant/edit/client.go index 4912f14e..3ed8f7e1 100644 --- a/server/core/internal/assistant/edit/client.go +++ b/server/core/internal/assistant/edit/client.go @@ -30,10 +30,10 @@ type Result struct { Errors []OpError `json:"errors"` } -// Client posts batched operations to the Node hocuspocus service's -// internal operations endpoint. The base URL points at the Node -// service (e.g. http://auth-realtime:8081); endpoint replaces its -// path with the per-document operations path. +// Client posts to the Node hocuspocus service's internal per-branch +// endpoints: batched operations and flushes. The base URL points at the +// Node service (e.g. http://auth-realtime:8081); endpoint replaces its +// path with the per-branch action path. type Client struct { // httpClient is the underlying HTTP client. Callers should // pass one with a reasonable timeout configured. @@ -44,8 +44,8 @@ type Client struct { } // NewClient constructs an edit-RPC client. The baseURL must include -// the scheme and host of the Node service; the operations path is -// appended per request. +// the scheme and host of the Node service; the action path is appended +// per request. func NewClient(httpClient *http.Client, baseURL string) *Client { if httpClient == nil { httpClient = http.DefaultClient @@ -102,7 +102,7 @@ func (c *Client) Apply( return Result{}, fmt.Errorf("marshaling operations: %w", err) } - endpoint, err := c.endpoint(documentID, branchID) + endpoint, err := c.endpoint(documentID, branchID, "operations") if err != nil { return Result{}, err } @@ -142,16 +142,50 @@ func (c *Client) Apply( return out, nil } -// endpoint builds the per-document operations URL. -func (c *Client) endpoint(documentID, branchID xid.ID) (string, error) { +// Flush has the Node service store what the editors of the branch hold +// and returns once core has it. +func (c *Client) Flush(ctx context.Context, documentID, branchID xid.ID) error { + endpoint, err := c.endpoint(documentID, branchID, "flush") + if err != nil { + return err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, http.NoBody) + if err != nil { + // NOCOV: the URL is pre-validated by endpoint, so the only + // remaining failure is a nil context. + return fmt.Errorf("building request: %w", err) + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("posting flush: %w", err) + } + defer resp.Body.Close() //nolint:errcheck // error provides no meaningful info + + if resp.StatusCode != http.StatusNoContent { + preview, err := io.ReadAll(io.LimitReader(resp.Body, _maxErrorPreviewBytes)) + if err != nil { + return fmt.Errorf("node flush endpoint: status %d: reading response body: %w", resp.StatusCode, err) + } + + return fmt.Errorf("node flush endpoint: status %d: %s", resp.StatusCode, string(preview)) + } + + return nil +} + +// endpoint builds the URL of a per-branch action. +func (c *Client) endpoint(documentID, branchID xid.ID, action string) (string, error) { base, err := url.Parse(c.baseURL) if err != nil { return "", fmt.Errorf("parsing base url: %w", err) } - base.Path = fmt.Sprintf("/api/internal/documents/%s/branches/%s/operations", + base.Path = fmt.Sprintf("/api/internal/documents/%s/branches/%s/%s", documentID, branchID, + action, ) return base.String(), nil diff --git a/server/core/internal/assistant/edit/client_test.go b/server/core/internal/assistant/edit/client_test.go index 256da848..e66c0b3a 100644 --- a/server/core/internal/assistant/edit/client_test.go +++ b/server/core/internal/assistant/edit/client_test.go @@ -235,6 +235,78 @@ func Test_Client_Apply(t *testing.T) { } } +func Test_Client_Flush(t *testing.T) { + t.Parallel() + + docID, branchID := xid.New(), xid.New() + path := "/api/internal/documents/" + docID.String() + "/branches/" + branchID.String() + "/flush" + + cc := map[string]struct { + BaseURL string + CloseEarly bool + StatusCode int + Body string + Err error + }{ + "Invalid base URL": { + BaseURL: "://bad", + Err: assert.AnError, + }, + "Transport error": { + CloseEarly: true, + Err: assert.AnError, + }, + "Non-204 status surfaces the body": { + StatusCode: http.StatusInternalServerError, + Body: `{"error":"pending changes could not be stored"}`, + Err: assert.AnError, + }, + "Successful flush": { + StatusCode: http.StatusNoContent, + }, + } + + for cn, c := range cc { + t.Run(cn, func(t *testing.T) { + t.Parallel() + + var gotPath, gotMethod string + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath, gotMethod = r.URL.Path, r.Method + + w.WriteHeader(c.StatusCode) + _, err := w.Write([]byte(c.Body)) + assert.NoError(t, err) + })) + defer srv.Close() + + base := srv.URL + if c.BaseURL != "" { + base = c.BaseURL + } + + if c.CloseEarly { + srv.Close() + } + + err := NewClient(srv.Client(), base).Flush(context.Background(), docID, branchID) + testutil.AssertEqualError(t, c.Err, err) + + if c.StatusCode == 0 { + return + } + + assert.Equal(t, path, gotPath) + assert.Equal(t, http.MethodPost, gotMethod) + + if c.Body != "" { + assert.Contains(t, err.Error(), c.Body) + } + }) + } +} + func Test_Client_endpoint(t *testing.T) { t.Parallel() @@ -259,7 +331,7 @@ func Test_Client_endpoint(t *testing.T) { t.Run(cn, func(t *testing.T) { t.Parallel() - got, err := NewClient(nil, c.BaseURL).endpoint(docID, branchID) + got, err := NewClient(nil, c.BaseURL).endpoint(docID, branchID, "operations") testutil.AssertEqualError(t, c.Err, err) assert.Equal(t, c.Expected, got) })