diff --git a/languages/golang/CHANGELOG.md b/languages/golang/CHANGELOG.md index 9d656a1aa..edf47afb5 100644 --- a/languages/golang/CHANGELOG.md +++ b/languages/golang/CHANGELOG.md @@ -77,6 +77,18 @@ listed under **Breaking**. Releases are tagged `languages/golang/vX.Y.Z`. ### Changed +- **A call whose context ends fails on its own and leaves the client + open.** A cancelled or timed-out call used to close the `encrypt.Client`, + or the `auth` profile behind its strategy, for every later caller, and a + request whose deadline passed while it waited for the client closed it + before running. Now a call waiting for the client leaves the queue with + the context's error; a ZeroKMS request in flight is cancelled, and the + error matches both the context's error and `ErrTransport`; a device + session's refresh in flight finishes in the background, for up to 10 + seconds, so a rotated refresh token is never lost; and the client stays + open. A client or profile still closes itself (`ErrState`) when the + guest traps, or when a call runs on for 30 seconds after its context + ended. - **stashgen refuses a match index on an index-only field** (`stash:"name,index=match"` with no `encrypt`). A match query's rows are checked by decrypting them, and such a field stores no ciphertext. diff --git a/languages/golang/auth/cancel_test.go b/languages/golang/auth/cancel_test.go new file mode 100644 index 000000000..1dda26166 --- /dev/null +++ b/languages/golang/auth/cancel_test.go @@ -0,0 +1,303 @@ +package auth + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cipherstash/stack/languages/golang/internal/guest" + "github.com/tetratelabs/wazero/api" +) + +// A profile serves every strategy, and through them every encrypt client, +// in a process: a request cancelled while its token is fetched must fail +// on its own and leave the profile open for the next. + +// tokenServer answers the access-key exchange, holding each request until +// the client gives up while hold is set. +func tokenServer(t *testing.T, hold *atomic.Bool) *httptest.Server { + t.Helper() + var jwt string + // The server notices a client that hung up a moment after the client + // has gone: it reads the closed connection on a goroutine of its own. + // So the test's end waits for each held request to see its + // cancellation, for a bounded time, before the server is closed (which + // would end the request's context itself and hide a request the client + // never cancelled). + var held sync.WaitGroup + done := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if hold.Load() { + held.Add(1) + defer held.Done() + // The server notices a client that hung up only once the body + // is read. + _, _ = io.Copy(io.Discard, r.Body) + select { + case <-r.Context().Done(): + return + case <-done: + } + select { + case <-r.Context().Done(): + case <-time.After(5 * time.Second): + t.Error("a held request was never cancelled") + } + return + } + fmt.Fprintf(w, `{"accessToken":%q,"expiry":%d}`, jwt, time.Now().Add(time.Hour).Unix()) + })) + t.Cleanup(server.Close) + t.Cleanup(func() { + close(done) + held.Wait() + }) + jwt = testJWT(t, server.URL) + return server +} + +func TestATokenFetchCancelledFailsOnlyItself(t *testing.T) { + guestOrSkip(t) + var hold atomic.Bool + server := tokenServer(t, &hold) + profile, err := Open(context.Background(), t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer profile.Close() + strategy, err := profile.AccessKey(context.Background(), testCRN, "CSAKtestKeyId.testKeySecret", WithBaseURL(server.URL)) + if err != nil { + t.Fatal(err) + } + defer strategy.Close() + + hold.Store(true) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + if _, err := strategy.Token(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("cancelled Token: %v, want the deadline", err) + } + hold.Store(false) + // The next Token waits for the abandoned one to finish. + if token, err := strategy.Token(context.Background()); err != nil || token == "" { + t.Fatalf("the next Token: %q, %v", token, err) + } + if profile.root.closed || profile.root.inst.module.IsClosed() { + t.Fatal("the profile was closed") + } +} + +// An access-key exchange rotates nothing, so it ends with its caller: the +// held request is cancelled at the caller's deadline, not run on for the +// refresh grace, and Close does not wait for it. +func TestAnAccessKeyExchangeEndsWithItsCaller(t *testing.T) { + guestOrSkip(t) + var hold atomic.Bool + server := tokenServer(t, &hold) + profile, err := Open(context.Background(), t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer profile.Close() + strategy, err := profile.AccessKey(context.Background(), testCRN, "CSAKtestKeyId.testKeySecret", WithBaseURL(server.URL)) + if err != nil { + t.Fatal(err) + } + + hold.Store(true) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + if _, err := strategy.Token(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("cancelled Token: %v, want the deadline", err) + } + // The grace is 10 seconds; a Close that waited for it would take that. + start := time.Now() + if err := strategy.Close(); err != nil { + t.Fatal(err) + } + if waited := time.Since(start); waited > 2*time.Second { + t.Fatalf("Close took %v: the exchange ran on past its caller", waited) + } +} + +// A device session's refresh is the request that rotates a credential: the +// service answers with a new refresh token and retires the old one. A +// caller that gives up returns at once, but the refresh runs on, and what +// it brings is saved. The next Token is served from it with no second +// request, and the profile on disk holds the rotated refresh token. +func TestARefreshAnsweredAfterItsCallerGaveUpIsKept(t *testing.T) { + guestOrSkip(t) + dir, _ := expiredDeviceProfile(t) + var requests atomic.Int32 + answered := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if requests.Add(1) > 1 { + // The old refresh token is retired once the service answers. + w.WriteHeader(http.StatusBadRequest) + fmt.Fprint(w, `{"error":"invalid_grant"}`) + return + } + _, _ = io.Copy(io.Discard, r.Body) + time.Sleep(150 * time.Millisecond) + fmt.Fprint(w, `{"access_token":"fresh","token_type":"Bearer","expires_in":3600,"refresh_token":"refresh-2"}`) + close(answered) + })) + t.Cleanup(server.Close) + profile, err := Open(context.Background(), dir) + if err != nil { + t.Fatal(err) + } + defer profile.Close() + ws, err := profile.WorkspaceStore(context.Background(), wsA) + if err != nil { + t.Fatal(err) + } + strategy, err := ws.DeviceSession(context.Background(), WithBaseURL(server.URL)) + if err != nil { + t.Fatal(err) + } + defer strategy.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + if _, err := strategy.Token(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("cancelled Token: %v, want the deadline", err) + } + select { + case <-answered: + t.Fatal("the cancelled Token waited for the service's answer") + default: + } + if token, err := strategy.Token(context.Background()); err != nil || token != "fresh" { + t.Fatalf("the next Token: %q, %v, want the refreshed token", token, err) + } + if n := requests.Load(); n != 1 { + t.Fatalf("%d refresh requests, want 1: the late answer was dropped", n) + } + data, err := os.ReadFile(filepath.Join(dir, "workspaces", wsA, "auth.json")) + if err != nil { + t.Fatal(err) + } + var stored map[string]any + if err := json.Unmarshal(data, &stored); err != nil { + t.Fatal(err) + } + if stored["refresh_token"] != "refresh-2" { + t.Fatalf("stored refresh token = %v, want the rotated one", stored["refresh_token"]) + } +} + +// A refresh runs on after its caller gives up, so the guest call under it +// can succeed after the caller's context ended. It still fails with the +// context's error; what the refresh brought is kept for the next call. +func TestARefreshThatSucceedsAfterItsContextEndsFailsWithIt(t *testing.T) { + guestOrSkip(t) + dir, _ := expiredDeviceProfile(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + cancel() + fmt.Fprint(w, `{"access_token":"fresh","token_type":"Bearer","expires_in":3600,"refresh_token":"refresh-2"}`) + })) + t.Cleanup(server.Close) + profile, err := Open(context.Background(), dir) + if err != nil { + t.Fatal(err) + } + defer profile.Close() + ws, err := profile.WorkspaceStore(context.Background(), wsA) + if err != nil { + t.Fatal(err) + } + strategy, err := ws.DeviceSession(context.Background(), WithBaseURL(server.URL)) + if err != nil { + t.Fatal(err) + } + defer strategy.Close() + + out, err := strategy.store.callArgs(withRefresh(ctx), func(i *instance) api.Function { return i.authRefresh }, guest.BufArg([]byte(strategy.handle))) + if !errors.Is(err, context.Canceled) || out != nil { + t.Fatalf("refresh: %q, %v, want no token and context.Canceled", out, err) + } + if token, err := strategy.Token(context.Background()); err != nil || token != "fresh" { + t.Fatalf("the next Token: %q, %v, want the refreshed token", token, err) + } +} + +func TestATokenRequestWhoseContextEndsInTheQueueNeverReachesTheGuest(t *testing.T) { + guestOrSkip(t) + var hold atomic.Bool + server := tokenServer(t, &hold) + profile, err := Open(context.Background(), t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer profile.Close() + strategy, err := profile.AccessKey(context.Background(), testCRN, "CSAKtestKeyId.testKeySecret", WithBaseURL(server.URL)) + if err != nil { + t.Fatal(err) + } + defer strategy.Close() + + for name, lock := range map[string]interface { + Acquire(context.Context) error + Unlock() + }{"the strategy's": &strategy.mu, "the profile's": &profile.root.mu} { + t.Run(name, func(t *testing.T) { + if err := lock.Acquire(context.Background()); err != nil { + t.Fatal(err) + } + released := make(chan struct{}) + time.AfterFunc(200*time.Millisecond, func() { lock.Unlock(); close(released) }) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + if _, err := strategy.Token(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("queued Token: %v, want the deadline", err) + } + <-released + }) + } + if token, err := strategy.Token(context.Background()); err != nil || token == "" { + t.Fatalf("the next Token: %q, %v", token, err) + } +} + +// The backstop: a call still running long after its caller gave up closes +// the profile. The caller has returned by then; the next call is ErrState. +func TestATokenFetchThatRunsOnAfterItsContextEndsClosesTheProfile(t *testing.T) { + guestOrSkip(t) + profile, err := Open(context.Background(), t.TempDir(), WithRoundTripper(roundTripFunc(func(*http.Request) (*http.Response, error) { + // A RoundTripper that ignores the request's context. + time.Sleep(300 * time.Millisecond) + return nil, errors.New("too late") + }))) + if err != nil { + t.Fatal(err) + } + defer profile.Close() + strategy, err := profile.AccessKey(context.Background(), testCRN, "CSAKtestKeyId.testKeySecret", WithBaseURL("https://cts.example.com")) + if err != nil { + t.Fatal(err) + } + defer strategy.Close() + profile.root.stuck = 20 * time.Millisecond + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + if _, err := strategy.Token(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("stuck Token: %v, want the deadline", err) + } + if _, err := strategy.Token(context.Background()); !errors.Is(err, ErrState) { + t.Fatalf("a Token after the close: %v, want ErrState", err) + } +} diff --git a/languages/golang/auth/doc.go b/languages/golang/auth/doc.go index b0bf4f16d..d61e99371 100644 --- a/languages/golang/auth/doc.go +++ b/languages/golang/auth/doc.go @@ -39,6 +39,18 @@ // [OAuth2TokenSource] adapts an existing golang.org/x/oauth2.TokenSource // into the OIDC provider interface. // +// A call whose context ends, [Strategy.Token] included, fails with the +// context's error and leaves the store and its strategies open: a request +// cancelled while its token is exchanged or refreshed does not stop the +// encrypt clients that read tokens from the same store. A request to the +// token service is cancelled with its caller, except a device session's +// refresh: the service rotates the refresh token when it answers one, so +// the caller returns at once while the refresh runs on, for up to 10 +// seconds, and the store keeps what it sends back for the next call. +// [Strategy.Close] waits for such a refresh. The store closes itself, and +// every later call is [ErrState], only when its guest traps or a call runs +// on for 30 seconds after its context ended. +// // # Why a second guest // // The crypto guest behind encrypt has no filesystem and no diff --git a/languages/golang/auth/guest.go b/languages/golang/auth/guest.go index 5c17ef261..69e395a3a 100644 --- a/languages/golang/auth/guest.go +++ b/languages/golang/auth/guest.go @@ -111,6 +111,9 @@ func newInstance(ctx context.Context, wasm []byte, hostDir string, policy guest. return nil, fmt.Errorf("%w: %s: %w", ErrNoProfile, hostDir, err) } } + // A call never runs under its caller's context (see guest.Detach), so + // this closes the module only for a call that ran on long after its + // caller gave up: ProfileStore.callArgs. config := wazero.NewRuntimeConfig(). WithCompilationCache(compilationCache()). WithCloseOnContextDone(true) @@ -183,8 +186,8 @@ func newInstance(ctx context.Context, wasm []byte, hostDir string, policy guest. // release runs the guest's shutdown — every buffer it still holds wiped — // and closes the runtime, which frees the linear memory through the -// allocator's wipe, then the directory handle the mount holds. A module an -// interrupted call or a trap already closed cannot run sa_shutdown; the +// allocator's wipe, then the directory handle the mount holds. A module a +// stuck call or a trap already closed cannot run sa_shutdown; the // runtime close still wipes and frees its memory. func (inst *instance) release() error { ctx := context.Background() diff --git a/languages/golang/auth/store.go b/languages/golang/auth/store.go index b5baea7a6..23b9b5915 100644 --- a/languages/golang/auth/store.go +++ b/languages/golang/auth/store.go @@ -10,7 +10,7 @@ import ( "path/filepath" "runtime" "strings" - "sync" + "time" "github.com/cipherstash/stack/languages/golang/internal/guest" "github.com/cipherstash/vitaminc/bindings/go/vcffi" @@ -53,6 +53,15 @@ func RequireLockedMemory() Option { // stores from one root share one guest instance over the one mounted // directory; Close on any of them releases it. It is safe for concurrent // use; calls are serialised, because a wasm instance is single-threaded. +// +// A call whose context ends fails with the context's error, and only that +// call: a request cancelled while its token is fetched or refreshed never +// closes the profile, or the strategies and encrypt clients that read +// tokens from it. A device session's refresh already sent finishes in the +// background, for up to 10 seconds, so a rotated refresh token is never +// lost (see refreshGrace). The profile closes itself, and every later call is +// [ErrState], only when the guest's state is unknown: it trapped, or it +// went on running long after its caller gave up, which is a bug. type ProfileStore struct { root *root // dir is the store's directory as the guest sees it: guestRoot, or a @@ -62,13 +71,17 @@ type ProfileStore struct { // root is the guest instance the stores of one profile share. type root struct { - mu sync.Mutex + mu guest.Lock inst *instance hostDir string - closed bool - // released is the runtime's own state, apart from closed: an - // interrupted call closes the module, and so the stores, while the - // runtime is still allocated. Close frees it even then. + // stuck is how long a call may run on after its context ends before + // the guest is treated as stuck and closed. guest.StuckAfter; tests + // shorten it. + stuck time.Duration + closed bool + // released is the runtime's own state, apart from closed: a stuck + // call closes the module, and so the stores, while the runtime is + // still allocated. Close frees it even then. released bool cleanup runtime.Cleanup } @@ -135,7 +148,7 @@ func open(ctx context.Context, hostDir string, opts []Option) (*ProfileStore, er if err != nil { return nil, err } - r := &root{inst: inst, hostDir: hostDir} + r := &root{inst: inst, hostDir: hostDir, stuck: guest.StuckAfter} // The cleanup takes the instance, not the root: a cleanup whose // argument reaches its object keeps that object alive forever. r.cleanup = runtime.AddCleanup(r, func(inst *instance) { _ = inst.release() }, inst) @@ -204,7 +217,7 @@ func (s *ProfileStore) Close() error { type export func(*instance) api.Function // call runs one export under the profile's lock, closing the profile if -// the guest trapped or an interrupted call took the module down. +// the guest trapped or a stuck call took the module down. func (s *ProfileStore) call(ctx context.Context, fn export, args ...string) ([]byte, error) { // Every profile export works on the mount; without one the guest would // report an I/O error that says less than this does. @@ -221,23 +234,32 @@ func (s *ProfileStore) call(ctx context.Context, fn export, args ...string) ([]b // callArgs is the shared checked call path for profile exports and auth // exports. The latter pass secret-bearing byte buffers and wipe host copies. +// +// The caller's context bounds the wait for the lock, as encrypt's +// Client.call does, and the guest runs under guest.Detach, so a caller +// that gives up never closes the profile. A refresh request runs on for +// refreshGrace, so the guest keeps the rotated token the service sends. func (s *ProfileStore) callArgs(ctx context.Context, fn export, args ...guest.Arg) ([]byte, error) { r := s.root - r.mu.Lock() + if err := r.mu.Acquire(ctx); err != nil { + return nil, err + } defer r.mu.Unlock() if r.closed || r.inst.module.IsClosed() { r.closed = true return nil, ErrState } growth := r.inst.mem.GrowthRefusal() - out, err := guest.Call(ctx, r.inst.mem, r.inst.module, r.inst.exports, fn(r.inst), args...) + run, done := guest.Detach(ctx, r.stuck) + out, err := guest.Call(run, r.inst.mem, r.inst.module, r.inst.exports, fn(r.inst), args...) + done() switch { case r.inst.module.IsClosed(): r.closed = true if err == nil { err = ErrState } - err = fmt.Errorf("%w: interrupted call closed the profile", err) + err = fmt.Errorf("%w: the guest ran on after its caller's context ended, and the profile is closed", err) case errors.Is(err, guest.ErrTrap): r.closed = true _ = r.inst.module.Close(context.Background()) @@ -253,8 +275,19 @@ func (s *ProfileStore) callArgs(ctx context.Context, fn export, args ...guest.Ar err = fmt.Errorf("%w (growth refused under RequireLockedMemory): %w", guest.MemoryLockError(g.Reason), err) } if err != nil { + if ctxErr := ctx.Err(); ctxErr != nil && !errors.Is(err, ctxErr) { + err = fmt.Errorf("%w: %w", ctxErr, err) + } return nil, err } + // A call that succeeded after its context ended fails too, as in + // encrypt's Client.call, and its output (a token, say) is wiped. What + // the guest wrote stays written: a refresh that ran on keeps the + // rotated token, and the next call reads it. + if ctxErr := ctx.Err(); ctxErr != nil { + guest.Wipe(out) + return nil, ctxErr + } return out, nil } diff --git a/languages/golang/auth/strategy.go b/languages/golang/auth/strategy.go index 5fadc7bce..51379cbdf 100644 --- a/languages/golang/auth/strategy.go +++ b/languages/golang/auth/strategy.go @@ -7,7 +7,6 @@ import ( "fmt" "os" "strconv" - "sync" "github.com/cipherstash/stack/languages/golang/internal/guest" "github.com/tetratelabs/wazero/api" @@ -58,7 +57,7 @@ type Strategy struct { handle string providerID uint32 device bool - mu sync.Mutex + mu guest.Lock closed bool } @@ -182,28 +181,63 @@ func (s *ProfileStore) hasToken(ctx context.Context) (bool, error) { // Token gets the current CTS bearer credential. Device sessions read a fresh // token without a file lock; only a refresh call takes the cross-process lock. func (s *Strategy) Token(ctx context.Context) (string, error) { - s.mu.Lock() - defer s.mu.Unlock() + // A request queued behind another's refresh leaves when its context + // ends, as at the profile's own lock. + if err := s.mu.Acquire(ctx); err != nil { + return "", err + } if s.closed { + s.mu.Unlock() return "", ErrState } - ctx, status := withAuthHTTPStatus(ctx) - token, err := s.token(ctx) - return token, status.wrap(err) + // The work runs on its own goroutine, which holds the lock until it is + // done, so a caller that gives up returns at once. A device session's + // refresh already sent finishes there, and the guest keeps what the + // service sends (see refreshGrace): a rotated refresh token is never + // lost. The token it brings is dropped; the next Token reads it. Any + // other request ends with the caller. + type result struct { + token string + err error + } + done := make(chan result, 1) + go func() { + defer s.mu.Unlock() + ctx, status := withAuthHTTPStatus(ctx) + token, err := s.token(ctx) + done <- result{token, status.wrap(err)} + }() + select { + case r := <-done: + return r.token, r.err + case <-ctx.Done(): + // A call that finished as the context ended still answers. + select { + case r := <-done: + return r.token, r.err + default: + return "", ctx.Err() + } + } } // token is Token under the strategy's lock: a read, then a locked refresh // for a device session that needs one. func (s *Strategy) token(ctx context.Context) (string, error) { - call := func(export func(*instance) api.Function) (string, error) { + call := func(ctx context.Context, export func(*instance) api.Function) (string, error) { out, err := s.store.callArgs(ctx, export, guest.BufArg([]byte(s.handle))) if err != nil { return "", err } defer guest.Wipe(out) + // A caller that has left gets no copy: a string cannot be wiped, + // and the next Token reads the guest's. + if err := ctx.Err(); err != nil { + return "", err + } return string(out), nil } - token, err := call(func(i *instance) api.Function { return i.authToken }) + token, err := call(ctx, func(i *instance) api.Function { return i.authToken }) if !s.device || !errors.Is(err, guest.ErrAuthRefreshRequired) { return token, err } @@ -213,7 +247,9 @@ func (s *Strategy) token(ctx context.Context) (string, error) { } err = withRefreshLock(ctx, path, func() error { var err error - token, err = call(func(i *instance) api.Function { return i.authRefresh }) + // The one request that rotates a credential: it runs on after its + // caller leaves (see refreshGrace). + token, err = call(withRefresh(ctx), func(i *instance) api.Function { return i.authRefresh }) return err }) return token, err @@ -229,6 +265,9 @@ func (s *Strategy) token(ctx context.Context) (string, error) { func (s *Strategy) MemoryLockError() error { return s.store.MemoryLockError() } // Close drops the strategy's cached token and unregisters its provider. +// It waits for a call in flight, including a device session's refresh that +// a cancelled Token left running: that refresh finishes, for up to 10 +// seconds, so the rotated refresh token it brings is saved rather than lost. func (s *Strategy) Close() error { s.mu.Lock() defer s.mu.Unlock() diff --git a/languages/golang/auth/transport.go b/languages/golang/auth/transport.go index 599f40dab..b2e18d89c 100644 --- a/languages/golang/auth/transport.go +++ b/languages/golang/auth/transport.go @@ -10,6 +10,7 @@ import ( "sort" "strings" "sync" + "time" "github.com/cipherstash/stack/languages/golang/internal/guest" "github.com/tetratelabs/wazero" @@ -70,7 +71,10 @@ func (s *authHTTPStatus) wrap(err error) error { } type authTransport struct { - rt http.RoundTripper + rt http.RoundTripper + // grace is how long a refresh request runs on after its caller gives + // up: refreshGrace, shortened by tests. + grace time.Duration mu sync.Mutex providers map[uint32]OIDCProvider nextProvider uint32 @@ -80,7 +84,7 @@ func newAuthTransport(rt http.RoundTripper) *authTransport { if rt == nil { rt = http.DefaultTransport } - return &authTransport{rt: rt, providers: make(map[uint32]OIDCProvider)} + return &authTransport{rt: rt, grace: refreshGrace, providers: make(map[uint32]OIDCProvider)} } func (t *authTransport) register(provider OIDCProvider) uint32 { @@ -101,6 +105,40 @@ func (t *authTransport) unregister(id uint32) { t.mu.Unlock() } +// refreshGrace is how long a device session's refresh request runs on after +// its caller's context ends. The token service rotates the refresh token +// when it answers a refresh, so a refresh abandoned after the service +// answered would lose the new token, and the next refresh would send one +// the service has retired. So a refresh is not cancelled with its caller: +// the caller returns at once (Strategy.Token), the request finishes, and +// the guest saves what came back. A refresh still running this long after +// its caller gave up fails as a transport failure, well inside +// guest.StuckAfter, so the profile stays open. +// +// Every other request to the token service (an access-key exchange, an +// OIDC exchange) rotates nothing, and is cancelled with its caller. +const refreshGrace = 10 * time.Second + +// refreshKey marks the context of a device session's refresh call +// (withRefresh), whose requests get refreshGrace. +type refreshKey struct{} + +// withRefresh marks ctx as a refresh call's. +func withRefresh(ctx context.Context) context.Context { + return context.WithValue(ctx, refreshKey{}, true) +} + +// requestContext is the context a request to the token service runs under: +// the caller's, so a caller that gives up cancels it, except for a refresh, +// which runs on for the grace (see refreshGrace). done must be called when +// the request is finished with. +func (t *authTransport) requestContext(ctx context.Context) (context.Context, func()) { + if refresh, _ := ctx.Value(refreshKey{}).(bool); refresh { + return guest.Detach(guest.Caller(ctx), t.grace) + } + return guest.Caller(ctx), func() {} +} + func (t *authTransport) instantiate(ctx context.Context, r wazero.Runtime) error { _, err := r.NewHostModuleBuilder("cipherstash_transport"). NewFunctionBuilder().WithFunc(t.send).Export("transport_send"). @@ -124,7 +162,9 @@ func (t *authTransport) send(ctx context.Context, m api.Module, return t.placeResponse(ctx, m, respHeadersPtrOut, respHeadersLenOut, respBodyPtrOut, respBodyLenOut, -1, nil, []byte("request buffer out of range")) } requestBody := newAuthRequestBody(body) - req, err := http.NewRequestWithContext(ctx, string(method), string(url), requestBody) + reqCtx, done := t.requestContext(ctx) + defer done() + req, err := http.NewRequestWithContext(reqCtx, string(method), string(url), requestBody) if err != nil { _ = requestBody.Close() return t.placeResponse(ctx, m, respHeadersPtrOut, respHeadersLenOut, respBodyPtrOut, respBodyLenOut, -1, nil, []byte(err.Error())) @@ -174,7 +214,7 @@ func (t *authTransport) oidcTokenGet(ctx context.Context, m api.Module, provider if source == nil { return 1 } - token, err := source.Token(ctx) + token, err := source.Token(guest.Caller(ctx)) if err != nil || token == "" || strings.ContainsAny(token, "\r\n\x00") { return 1 } diff --git a/languages/golang/encrypt/README.md b/languages/golang/encrypt/README.md index fe6021930..79ff551ba 100644 --- a/languages/golang/encrypt/README.md +++ b/languages/golang/encrypt/README.md @@ -117,6 +117,25 @@ ZeroKMS round trip `NewClient` makes. It has nothing to do with an *encryption* context, which is what a field is sealed under; that is the `context=` tag. +Every call takes a `ctx` too, and a call whose `ctx` is cancelled or passes +its deadline fails with the context's error. Only that call fails: one +request giving up never closes the shared client for the others. + +- A call waiting for the client (it runs one call at a time) leaves the + queue when its `ctx` ends, and never starts. +- A call waiting on ZeroKMS has that request cancelled. Its error matches + both `context.Canceled` or `context.DeadlineExceeded` and `ErrTransport`. +- A call waiting on its auth strategy for a token returns too, with the + context's error. If the strategy is refreshing a device session, the + refresh finishes in the background, for up to 10 seconds. The service + rotates the refresh token when it answers, and dropping that answer + would leave the strategy unable to refresh. + +A client closes itself, and every later call returns `ErrState`, only when +the guest's state is unknown: the guest trapped, or a call ran on for 30 +seconds after its `ctx` ended, which means a bug. A service that sees +`ErrState` from a client it did not close builds a new one. + Every other setting is a functional option with a default: ```go @@ -372,7 +391,11 @@ or for stored data. Check an error two ways. its field's context (ZeroKMS refuses the key, or the AEAD fails). - `ErrEncoding`: a stored value or a call that does not fit the declaration. - `ErrUnauthorized`, `ErrNotFound`, `ErrTransport`, `ErrKMS`: ZeroKMS. -- `ErrState`: a call on a closed client. `ErrMemoryLock`: see below. +- `ErrState`: a call on a closed client: you closed it, or the guest + trapped or ran on long after a call's context ended (see Connect). + `ErrMemoryLock`: see below. +- `context.Canceled`, `context.DeadlineExceeded`: the call's `ctx` ended. + The client is still open. `errors.As` with a `*Diagnostic` gives you the detail behind a failure the engine reports: a stable `Code` (`stack_encrypt::foreign_keyset`, diff --git a/languages/golang/encrypt/cancel_test.go b/languages/golang/encrypt/cancel_test.go new file mode 100644 index 000000000..5acaae70e --- /dev/null +++ b/languages/golang/encrypt/cancel_test.go @@ -0,0 +1,393 @@ +package encrypt + +import ( + "context" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "math/rand/v2" + "net/http" + "net/http/httptest" + "runtime" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cipherstash/stack/languages/golang/auth" + "github.com/cipherstash/stack/languages/golang/internal/guest" + "github.com/cipherstash/stack/languages/golang/internal/record" +) + +// One encrypt.Client serves every request in a service, so a caller's +// context ending must fail that caller's call and nothing else. These tests +// pin that contract, and the one case that still closes the client: a guest +// that runs on long after its caller gave up. + +// unauthorized answers every ZeroKMS request with a 401: a full round trip +// through the guest's network path that ends in a status the client can +// tell apart from ErrState. +func unauthorized(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusUnauthorized, Body: io.NopCloser(strings.NewReader("nope")), Header: http.Header{}}, nil +} + +// switchable is a RoundTripper whose behaviour a test changes between calls. +type switchable struct { + mu sync.Mutex + rt roundTripFunc + sent atomic.Int32 +} + +func (s *switchable) set(rt roundTripFunc) { + s.mu.Lock() + defer s.mu.Unlock() + s.rt = rt +} + +func (s *switchable) RoundTrip(r *http.Request) (*http.Response, error) { + s.sent.Add(1) + s.mu.Lock() + rt := s.rt + s.mu.Unlock() + return rt(r) +} + +// cancelClient is a guest not yet given a key, whose ZeroKMS requests go to +// rt and whose token comes from token. Its se_cipher_init is the call the +// tests drive: it makes a load-keyset request, so it reaches the network. +func cancelClient(t *testing.T, rt http.RoundTripper, token tokenSource) *Client { + t.Helper() + tr := &transport{rt: rt, token: token} + inst, err := newInstance(context.Background(), guestOrSkip(t), tr, guest.BestEffort) + if err != nil { + t.Fatal(err) + } + c := newClient(inst, tr) + t.Cleanup(func() { _ = c.Close() }) + return c +} + +// cipherInit runs se_cipher_init on c under ctx. +func cipherInit(ctx context.Context, c *Client) error { + encoded, err := encodeConfig(testInit("http://zerokms.invalid")) + if err != nil { + return err + } + defer wipe(encoded) + _, err = c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { + return inst.call(ctx, inst.cipherInit, buf(encoded)) + }) + return err +} + +// wantOpen fails the test unless c is still open: its next call reaches +// ZeroKMS and comes back with ZeroKMS's answer, not ErrState. +func wantOpen(t *testing.T, c *Client, rt *switchable) { + t.Helper() + if c.closed || c.inst.module.IsClosed() { + t.Fatal("the client was closed") + } + rt.set(unauthorized) + if err := cipherInit(context.Background(), c); !errors.Is(err, ErrUnauthorized) { + t.Fatalf("the next call: %v, want ErrUnauthorized from ZeroKMS", err) + } +} + +func TestACallCancelledDuringARequestFailsOnlyItself(t *testing.T) { + rt := &switchable{rt: func(r *http.Request) (*http.Response, error) { + <-r.Context().Done() + return nil, r.Context().Err() + }} + c := cancelClient(t, rt, staticToken("t")) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + err := cipherInit(ctx, c) + if !errors.Is(err, context.DeadlineExceeded) || !errors.Is(err, ErrTransport) { + t.Fatalf("cancelled call: %v, want the deadline and ErrTransport", err) + } + wantOpen(t, c, rt) +} + +func TestACallCancelledDuringATokenFetchFailsOnlyItself(t *testing.T) { + rt := &switchable{rt: unauthorized} + var blocking atomic.Bool + blocking.Store(true) + token := tokenFunc(func(ctx context.Context) (string, error) { + if blocking.Load() { + <-ctx.Done() + return "", ctx.Err() + } + return "t", nil + }) + c := cancelClient(t, rt, token) + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(50*time.Millisecond, cancel) + err := cipherInit(ctx, c) + if !errors.Is(err, context.Canceled) || !errors.Is(err, ErrTransport) { + t.Fatalf("cancelled call: %v, want context.Canceled and ErrTransport", err) + } + if n := rt.sent.Load(); n != 0 { + t.Fatalf("%d request(s) sent without a token", n) + } + blocking.Store(false) + wantOpen(t, c, rt) +} + +func TestACallWhoseContextEndsInTheQueueNeverReachesTheGuest(t *testing.T) { + rt := &switchable{rt: unauthorized} + c := cancelClient(t, rt, staticToken("t")) + + t.Run("ends while waiting", func(t *testing.T) { + // Another call holds the client: the slow ZeroKMS round trip the + // queue is waiting behind. + if err := c.mu.Acquire(context.Background()); err != nil { + t.Fatal(err) + } + released := make(chan struct{}) + time.AfterFunc(200*time.Millisecond, func() { c.mu.Unlock(); close(released) }) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + err := cipherInit(ctx, c) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("queued call: %v, want the deadline", err) + } + // It left at its deadline, not when the lock came free. + select { + case <-released: + t.Error("the queued call waited past its deadline for the lock") + default: + } + <-released + }) + t.Run("already ended", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := cipherInit(ctx, c); !errors.Is(err, context.Canceled) { + t.Fatalf("call with an ended context: %v, want context.Canceled", err) + } + }) + if n := rt.sent.Load(); n != 0 { + t.Fatalf("%d request(s) reached ZeroKMS from calls whose context had ended", n) + } + wantOpen(t, c, rt) +} + +// The backstop: a call still running long after its caller gave up is a +// guest or host function that ignores cancellation. Its state is unknown, +// so the client closes, and says so. +func TestACallThatRunsOnAfterItsContextEndsClosesTheClient(t *testing.T) { + rt := &switchable{rt: func(*http.Request) (*http.Response, error) { + // A RoundTripper that ignores the request's context. + time.Sleep(300 * time.Millisecond) + return unauthorized(nil) + }} + c := cancelClient(t, rt, staticToken("t")) + c.stuck = 20 * time.Millisecond + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + err := cipherInit(ctx, c) + if err == nil || !strings.Contains(err.Error(), "the client is closed") { + t.Fatalf("stuck call: %v, want the client closed", err) + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("stuck call: %v, want it to match the deadline", err) + } + if err := cipherInit(context.Background(), c); !errors.Is(err, ErrState) { + t.Fatalf("a call after the close: %v, want ErrState", err) + } +} + +// A call that finishes in time stops the backstop: the client stays open +// however long after the call its context ends. +func TestAFinishedCallStopsTheBackstop(t *testing.T) { + rt := &switchable{rt: unauthorized} + c := cancelClient(t, rt, staticToken("t")) + c.stuck = time.Millisecond + ctx, cancel := context.WithCancel(context.Background()) + if err := cipherInit(ctx, c); !errors.Is(err, ErrUnauthorized) { + t.Fatalf("call: %v, want ErrUnauthorized", err) + } + cancel() + time.Sleep(20 * time.Millisecond) + wantOpen(t, c, rt) +} + +// Work the guest does without the network runs on through a cancellation, +// so a call can succeed after its caller gave up. It still fails with the +// context's error, and the output nobody will read is wiped. +func TestACallThatSucceedsAfterItsContextEndsFailsWithIt(t *testing.T) { + rt := &switchable{rt: unauthorized} + c := cancelClient(t, rt, staticToken("t")) + ctx, cancel := context.WithCancel(context.Background()) + plaintext := []byte("plaintext") + out, err := c.call(ctx, func(context.Context, *instance) ([]byte, error) { + cancel() + return plaintext, nil + }) + if !errors.Is(err, context.Canceled) || out != nil { + t.Fatalf("call: %q, %v, want no output and context.Canceled", out, err) + } + if string(plaintext) != string(make([]byte, len(plaintext))) { + t.Error("the discarded output was not wiped") + } + wantOpen(t, c, rt) +} + +// A closed module is ErrState, and Close still releases the runtime. +func TestAClosedModuleIsState(t *testing.T) { + ctx := context.Background() + c := rawInstance(t) + if err := c.inst.module.Close(ctx); err != nil { + t.Fatal(err) + } + if _, err := c.Keyset(KeysetName("k")).KeysetID(ctx); !errors.Is(err, ErrState) { + t.Fatalf("Keyset on a closed module: %v, want ErrState", err) + } + if err := c.Close(); err != nil { + t.Fatalf("Close after the module closed: %v", err) + } + // The close must reach the runtime. A stuck call closes the module and + // marks the client closed; a Close that treated that as "already done" + // would leave the runtime and the host module it carries allocated for + // the life of the process. + if !c.released { + t.Error("Close after the module closed left the runtime unreleased") + } + if m := c.inst.runtime.Module(transportModule); m != nil { + t.Errorf("host module %s is still registered after Close", transportModule) + } + // Still idempotent. + if err := c.Close(); err != nil { + t.Fatalf("second Close: %v", err) + } +} + +// Many callers share one client, as a service's requests do, and each call +// gets a deadline: none, already passed, or one that may pass in the queue +// or in the guest. The client never closes, every call whose context did +// not end succeeds, and nothing is left running afterwards. +func TestManyCallersCancellingAtRandomShareOneClient(t *testing.T) { + c, err := NewDeterministicClient(context.Background(), [32]byte{7}) + if errors.Is(err, ErrDeterministicGuestNotBuilt) { + t.Skip(err) + } + if err != nil { + t.Fatal(err) + } + defer c.Close() + cipher := c.DefaultKeyset() + plan := usersPlan() + before := runtime.NumGoroutine() + + const callers, calls = 16, 40 + var succeeded atomic.Int32 + var wg sync.WaitGroup + for g := range callers { + wg.Add(1) + go func() { + defer wg.Done() + r := rand.New(rand.NewPCG(uint64(g), 1)) + for i := range calls { + ctx, cancel := context.Background(), context.CancelFunc(func() {}) + switch r.IntN(4) { + case 0: // no deadline + case 1: + ctx, cancel = context.WithCancel(ctx) + cancel() + default: + ctx, cancel = context.WithTimeout(ctx, time.Duration(r.IntN(2000))*time.Microsecond) + } + rows := []record.Source{usersRow(uint32(i), "user@example.com")} + _, err := cipher.Seal(ctx, plan, rows) + ended := ctx.Err() != nil + cancel() + switch { + case errors.Is(err, ErrState): + t.Errorf("the client closed: %v", err) + return + case err == nil: + succeeded.Add(1) + case !ended: + t.Errorf("a call whose context did not end failed: %v", err) + case !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded): + t.Errorf("a cancelled call: %v, want the context's error", err) + } + } + }() + } + wg.Wait() + if succeeded.Load() == 0 { + t.Fatal("no call succeeded") + } + if _, err := cipher.Seal(context.Background(), plan, []record.Source{usersRow(1, "a@example.com")}); err != nil { + t.Fatalf("a call after the load: %v", err) + } + // Every backstop timer and context watcher has stopped. + deadline := time.Now().Add(2 * time.Second) + for runtime.NumGoroutine() > before && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if n := runtime.NumGoroutine(); n > before { + t.Errorf("%d goroutines after the load, %d before", n, before) + } +} + +// The encrypt client and the auth profile fail together or not at all: a +// call cancelled while the auth strategy exchanges its access key for a +// token leaves both open, and the next call gets a token and reaches +// ZeroKMS. +func TestACallCancelledDuringATokenExchangeLeavesTheProfileOpen(t *testing.T) { + var hold atomic.Bool + done := make(chan struct{}) + var jwt string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + if hold.Load() { + select { + case <-r.Context().Done(): + case <-done: + t.Error("a held token request was never cancelled") + } + return + } + fmt.Fprintf(w, `{"accessToken":%q,"expiry":%d}`, jwt, time.Now().Add(time.Hour).Unix()) + })) + t.Cleanup(server.Close) + t.Cleanup(func() { close(done) }) + claims, err := json.Marshal(map[string]any{"iss": server.URL, "sub": "CS|test", "workspace": "ZVATKW3VHMFG27DY", "exp": time.Now().Add(time.Hour).Unix()}) + if err != nil { + t.Fatal(err) + } + jwt = "e30." + base64.RawURLEncoding.EncodeToString(claims) + ".c2ln" + + store, err := auth.OpenWithoutProfile(context.Background()) + if errors.Is(err, auth.ErrGuestNotBuilt) { + t.Skip(err) + } + if err != nil { + t.Fatal(err) + } + defer store.Close() + strategy, err := store.AccessKey(context.Background(), testCRN, testAccessKey, auth.WithBaseURL(server.URL)) + if err != nil { + t.Fatal(err) + } + defer strategy.Close() + + rt := &switchable{rt: unauthorized} + c := cancelClient(t, rt, strategy) + hold.Store(true) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + if err := cipherInit(ctx, c); !errors.Is(err, context.DeadlineExceeded) || !errors.Is(err, ErrTransport) { + t.Fatalf("call cancelled during the exchange: %v, want the deadline and ErrTransport", err) + } + hold.Store(false) + wantOpen(t, c, rt) + if n := rt.sent.Load(); n != 1 { + t.Fatalf("%d ZeroKMS request(s), want the one after the token arrived", n) + } +} diff --git a/languages/golang/encrypt/checker.go b/languages/golang/encrypt/checker.go index 38533b0c2..a4ce1228d 100644 --- a/languages/golang/encrypt/checker.go +++ b/languages/golang/encrypt/checker.go @@ -57,7 +57,7 @@ func (k *Checker) engineCheck(ctx context.Context, plan *record.Plan) error { if err != nil { return fmt.Errorf("%w: %v", ErrEncoding, err) } - _, err = k.c.call(ctx, func(inst *instance) ([]byte, error) { + _, err = k.c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.planCheck, buf(encoded)) }) return err @@ -66,7 +66,7 @@ func (k *Checker) engineCheck(ctx context.Context, plan *record.Plan) error { // Targets lists the EQL types this build of the engine produces: none until // the EQL target dispatch lands. func (k *Checker) Targets(ctx context.Context) ([]record.Target, error) { - out, err := k.c.call(ctx, func(inst *instance) ([]byte, error) { + out, err := k.c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.targets) }) if err != nil { @@ -93,7 +93,7 @@ func (k *Checker) Operators(ctx context.Context, plan *record.Plan, field string if err != nil { return nil, fmt.Errorf("%w: %v", ErrEncoding, err) } - out, err := k.c.call(ctx, func(inst *instance) ([]byte, error) { + out, err := k.c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.operators, buf(encoded), buf([]byte(field))) }) if err != nil { diff --git a/languages/golang/encrypt/client.go b/languages/golang/encrypt/client.go index 4f89bd4ff..b0156425a 100644 --- a/languages/golang/encrypt/client.go +++ b/languages/golang/encrypt/client.go @@ -8,7 +8,7 @@ import ( "net/http" "runtime" "strconv" - "sync" + "time" "github.com/cipherstash/stack/languages/golang/internal/guest" "github.com/cipherstash/vitaminc/bindings/go/vcffi" @@ -20,6 +20,15 @@ import ( // concurrent use; calls are serialised internally, because a wasm instance // is single-threaded. Close it when done, as with any resource. // +// A call whose context ends fails with the context's error, and only that +// call: one request's cancellation or deadline never closes the client for +// the others sharing it. A call waiting its turn leaves the queue when its +// context ends; a call waiting on ZeroKMS has its request cancelled, and +// the error matches both the context's error and [ErrTransport]. The +// client closes itself, and every later call is [ErrState], only when the +// guest's state is unknown: it trapped, or it went on running long after +// its caller gave up, which is a bug. +// // Its key material lives in the guest's linear memory, which this package // supplies: reserved once so it never moves, locked in RAM and excluded // from core dumps where the platform allows, and wiped before it is @@ -29,13 +38,17 @@ import ( // where the protection lives. [Client.MemoryLocked] reports whether the // lock was granted. type Client struct { - mu sync.Mutex + mu guest.Lock inst *instance transport *transport - // closed refuses further calls: either Close ran, or an interrupted - // call took the module down under us (see Client.call). released is - // the runtime's own state, tracked apart from it because those two - // things come apart: an interrupted call closes the module — and so + // stuck is how long a call may run on after its context ends before + // the guest is treated as stuck and closed. guest.StuckAfter; tests + // shorten it. + stuck time.Duration + // closed refuses further calls: either Close ran, or a stuck or + // trapped call took the module down under us (see Client.call). + // released is the runtime's own state, tracked apart from it because + // those two things come apart: a stuck call closes the module — and so // the client — while the runtime and the host modules beside it are // still allocated. Close is what frees those, so it must do its work // even on a client that is already closed. @@ -175,7 +188,7 @@ func NewClient(ctx context.Context, opts ...ClientOption) (_ *Client, err error) c.releaseCredentials = resolved.Close c.credentialsLockErr = resolved.MemoryLockError owned = true - out, err := c.call(ctx, func(inst *instance) ([]byte, error) { + out, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.cipherInit, buf(encoded)) }) if err != nil { @@ -217,7 +230,7 @@ func consumeUnresolved(creds Credentials) { // instance, not the client: a cleanup whose argument reaches its object // keeps that object alive forever. func newClient(inst *instance, t *transport) *Client { - c := &Client{inst: inst, transport: t} + c := &Client{inst: inst, transport: t, stuck: guest.StuckAfter} c.cleanup = runtime.AddCleanup(c, func(inst *instance) { _ = inst.release() }, inst) return c } @@ -323,8 +336,8 @@ func encodeConfig(cfg initConfig) ([]byte, error) { func (c *Client) Close() error { c.mu.Lock() defer c.mu.Unlock() - // Idempotency turns on the runtime, not on the client: a client an - // interrupted call already closed has never released its runtime. + // Idempotency turns on the runtime, not on the client: a client a + // stuck call already closed has never released its runtime. if c.released { return nil } @@ -368,7 +381,7 @@ func (c *Client) resolveKeyset(ctx context.Context, sel KeysetSelector) (KeysetI if err != nil { return KeysetID{}, err } - out, err := c.call(ctx, func(inst *instance) ([]byte, error) { + out, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.keyset, buf(encoded)) }) if err != nil { @@ -382,18 +395,27 @@ func (c *Client) resolveKeyset(ctx context.Context, sel KeysetSelector) (KeysetI return id, nil } -// call runs f on the instance under the client's lock. +// call runs f on the instance under the client's lock. f must run its +// guest call under the context it is given, never the caller's. // -// A call interrupted by its context (the runtime closes the module on a -// deadline or cancellation, see newInstance) leaves the instance closed: -// its key material is gone with its memory and no further call can run. -// The client is then closed, so later calls are ErrState rather than a -// runtime error, and Close releases the runtime without a shutdown call. -// A guest that trapped is closed the same way, by this method: the guest -// builds with panic-as-abort, so a trap is an abort mid-export, after -// which its state is unknown and its keys are better wiped than reused. -func (c *Client) call(ctx context.Context, f func(*instance) ([]byte, error)) ([]byte, error) { - c.mu.Lock() +// The caller's context bounds the wait for the lock, and is refused if it +// has already ended, so a request that timed out in the queue never +// reaches the guest. The guest then runs under guest.Detach: the caller's +// context reaches the host functions, where it cancels a ZeroKMS request +// or a token fetch, and the guest returns through its ordinary error path +// with the client intact. The error matches the context's error too. +// +// The client is closed, so later calls are ErrState rather than a runtime +// error, when the guest's state is unknown. A guest that trapped is one: +// it builds with panic-as-abort, so a trap is an abort mid-export, after +// which its keys are better wiped than reused. A guest still running +// c.stuck after its caller's context ended is the other: Detach's context +// ends, the runtime closes the module (WithCloseOnContextDone, see +// newInstance), and its key material goes with its memory. +func (c *Client) call(ctx context.Context, f func(context.Context, *instance) ([]byte, error)) ([]byte, error) { + if err := c.mu.Acquire(ctx); err != nil { + return nil, err + } defer c.mu.Unlock() if c.closed || c.inst.module.IsClosed() { c.closed = true @@ -403,14 +425,16 @@ func (c *Client) call(ctx context.Context, f func(*instance) ([]byte, error)) ([ if c.transport != nil { c.transport.tokenErr = nil } - out, err := f(c.inst) + run, done := guest.Detach(ctx, c.stuck) + out, err := f(run, c.inst) + done() switch { case c.inst.module.IsClosed(): c.closed = true if err == nil { err = ErrState } - err = fmt.Errorf("%w: interrupted call closed the client", err) + err = fmt.Errorf("%w: the guest ran on after its caller's context ended, and the client is closed", err) case errors.Is(err, errGuestTrap): // No guest code is running (f has returned), so the close is // immediate: the module's memory is wiped and freed here. @@ -438,8 +462,23 @@ func (c *Client) call(ctx context.Context, f func(*instance) ([]byte, error)) ([ err = fmt.Errorf("%w (token source: %w)", err, c.transport.tokenErr) } } + // A call that failed after its context ended failed because of it, or + // as good as: the caller asks errors.Is(err, context.Canceled) or + // DeadlineExceeded, and the cause (ErrTransport for a cancelled + // request) stays matchable too. if err != nil { + if ctxErr := ctx.Err(); ctxErr != nil && !errors.Is(err, ctxErr) { + err = fmt.Errorf("%w: %w", ctxErr, err) + } return nil, err } + // A call that succeeded after its context ended fails too: work the + // guest does without the network (a query term, a batch under a + // cached key) runs on through a cancellation, and the caller was + // promised the context's error. The output nobody will read is wiped. + if ctxErr := ctx.Err(); ctxErr != nil { + wipe(out) + return nil, ctxErr + } return out, nil } diff --git a/languages/golang/encrypt/diagnostic_test.go b/languages/golang/encrypt/diagnostic_test.go index 811496d44..c632bde94 100644 --- a/languages/golang/encrypt/diagnostic_test.go +++ b/languages/golang/encrypt/diagnostic_test.go @@ -36,7 +36,7 @@ func wantCode(t *testing.T, err error, code string) { func TestGuestRefusalsCarryTheirDiagnostic(t *testing.T) { ctx := context.Background() c := rawInstance(t) - _, err := c.call(ctx, func(inst *instance) ([]byte, error) { + _, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.planCheck, buf([]byte{0xff})) }) if !errors.Is(err, ErrEncoding) { @@ -48,7 +48,7 @@ func TestGuestRefusalsCarryTheirDiagnostic(t *testing.T) { if err != nil { t.Fatal(err) } - _, err = c.call(ctx, func(inst *instance) ([]byte, error) { + _, err = c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.keyset, buf(selector)) }) if !errors.Is(err, ErrState) { @@ -65,7 +65,7 @@ func TestAGuestWithoutLastErrorFailsWithTheBareSentinel(t *testing.T) { ctx := context.Background() c := rawInstance(t) c.inst.exports.LastError = nil - _, err := c.call(ctx, func(inst *instance) ([]byte, error) { + _, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.planCheck, buf([]byte{0xff})) }) if err != ErrEncoding { diff --git a/languages/golang/encrypt/doc.go b/languages/golang/encrypt/doc.go index 6fc832d00..3650673b9 100644 --- a/languages/golang/encrypt/doc.go +++ b/languages/golang/encrypt/doc.go @@ -184,6 +184,27 @@ // memory until the garbage collector reclaims it; a MatchQuery never prints // it. // +// # Cancellation +// +// Every call takes a [context.Context], and a call whose context is +// cancelled or passes its deadline fails with that context's error. Only +// that call fails: one Client is shared by every request in a service, and +// one request giving up never closes it for the others. A call waiting +// for the client, which runs one call at a time, leaves the queue when its +// context ends and never starts. A call waiting on ZeroKMS has that request +// cancelled; its error matches both the context's error and +// [ErrTransport]. A call waiting on its auth strategy for a token returns +// too, with the context's error. If the strategy is refreshing a device +// session, the refresh finishes in the background, for up to 10 seconds: +// the service rotates the refresh token when it answers, and dropping that +// answer would leave the strategy unable to refresh. +// +// A Client closes itself, and every later call returns [ErrState], only +// when the guest's state is unknown: the guest trapped, or a call ran on +// for 30 seconds after its context ended, which means a bug, not a caller +// that gave up. A service that sees ErrState from a client it did not +// close builds a new one. +// // # Transport and auth // // The guest imports exactly two host functions: an HTTP send, served by any @@ -216,7 +237,8 @@ // // The generator and the compiler find a mistake in a declaration, so no call // returns an error for one. A call returns an error for a key, for the -// network, or for stored data: [ErrForeignKeyset] when a *Cipher is given a +// network, or for stored data, or because its context ended (see +// Cancellation): [ErrForeignKeyset] when a *Cipher is given a // record another keyset sealed, [ErrAuthentication] or [ErrForbidden] for a // ciphertext that does not open under its field's context, [ErrEncoding] for // a stored value that does not fit its declaration. Read the kind with diff --git a/languages/golang/encrypt/export_test.go b/languages/golang/encrypt/export_test.go index 8fd779c95..753d1a196 100644 --- a/languages/golang/encrypt/export_test.go +++ b/languages/golang/encrypt/export_test.go @@ -118,7 +118,7 @@ func deterministicClient(ctx context.Context, wasm []byte, seed [32]byte) (*Clie _ = c.Close() return nil, err } - out, err := c.call(ctx, func(inst *instance) ([]byte, error) { + out, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.cipherInit, buf(encoded)) }) if err != nil { diff --git a/languages/golang/encrypt/guest.go b/languages/golang/encrypt/guest.go index def93995e..2cffd3be2 100644 --- a/languages/golang/encrypt/guest.go +++ b/languages/golang/encrypt/guest.go @@ -113,10 +113,13 @@ func guestModuleConfig() wazero.ModuleConfig { // strict policy, memory that cannot be locked fails instantiation with // ErrMemoryLock. func newInstance(ctx context.Context, wasm []byte, t *transport, policy guest.LockPolicy) (*instance, error) { - // WithCloseOnContextDone lets a caller's deadline or cancellation - // interrupt an in-flight guest call — which otherwise holds the Client's - // lock against every other user. An interrupted call closes the module, - // so the Client is done afterwards; the alternative is a wedged process. + // WithCloseOnContextDone closes the module when a call's context ends. + // A guest call never runs under the caller's context: Client.call runs + // it under guest.Detach, whose context ends only when the guest has run + // on long after its caller gave up. Closing the module then frees the + // Client's lock, which a stuck call would otherwise hold against every + // other user, at the cost of the Client: the alternative is a wedged + // process. config := wazero.NewRuntimeConfig(). WithCompilationCache(compilationCache()). WithCloseOnContextDone(true) @@ -196,8 +199,8 @@ func newInstance(ctx context.Context, wasm []byte, t *transport, policy guest.Lo // release runs the guest's shutdown — the client key and every loaded // index key wiped inside the instance — and closes the runtime, which // frees the linear memory through the allocator's wipe. It is what Close -// does, and what the cleanup on an unreachable Client does. A module an -// interrupted call or a trap already closed cannot run se_shutdown; the +// does, and what the cleanup on an unreachable Client does. A module a +// stuck call or a trap already closed cannot run se_shutdown; the // runtime close still wipes and frees its memory, so nothing is left // behind either way. func (inst *instance) release() error { diff --git a/languages/golang/encrypt/guest_test.go b/languages/golang/encrypt/guest_test.go index a98266f3b..9981d753e 100644 --- a/languages/golang/encrypt/guest_test.go +++ b/languages/golang/encrypt/guest_test.go @@ -15,13 +15,11 @@ import ( "strconv" "strings" "testing" - "time" "github.com/cipherstash/stack/languages/golang/auth" "github.com/cipherstash/stack/languages/golang/internal/guest" "github.com/cipherstash/stack/languages/golang/internal/record" "github.com/tetratelabs/wazero" - "github.com/tetratelabs/wazero/sys" ) // Tests that drive the embedded guest without a live ZeroKMS. What they @@ -278,53 +276,6 @@ type roundTripFunc func(*http.Request) (*http.Response, error) func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } -// An interrupted call closes the module (WithCloseOnContextDone); the -// client must then be closed rather than a wedge or a runtime error. -func TestInterruptedCallClosesTheClient(t *testing.T) { - t.Run("deadline during a request", func(t *testing.T) { - guestOrSkip(t) - cfg := testConfig("http://zerokms.invalid") - cfg = append(cfg, WithTransport(roundTripFunc(func(r *http.Request) (*http.Response, error) { - <-r.Context().Done() - return nil, r.Context().Err() - }))) - ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) - defer cancel() - _, err := NewClient(ctx, cfg...) - if !errors.Is(err, context.DeadlineExceeded) { - t.Fatalf("NewClient: %v, want the deadline", err) - } - }) - t.Run("closed module is ErrState", func(t *testing.T) { - ctx := context.Background() - c := rawInstance(t) - // What the runtime does to the module when a call's context ends. - if err := c.inst.module.CloseWithExitCode(ctx, sys.ExitCodeContextCanceled); err != nil { - t.Fatal(err) - } - if _, err := c.Keyset(KeysetName("k")).KeysetID(ctx); !errors.Is(err, ErrState) { - t.Fatalf("Keyset on a closed module: %v, want ErrState", err) - } - if err := c.Close(); err != nil { - t.Fatalf("Close after interruption: %v", err) - } - // The close must reach the runtime. An interrupted call closes the - // module and marks the client closed; a Close that treated that as - // "already done" would leave the runtime and the host module it - // carries allocated for the life of the process. - if !c.released { - t.Error("Close after interruption left the runtime unreleased") - } - if m := c.inst.runtime.Module(transportModule); m != nil { - t.Errorf("host module %s is still registered after Close", transportModule) - } - // Still idempotent. - if err := c.Close(); err != nil { - t.Fatalf("second Close: %v", err) - } - }) -} - // A response the host would have to buffer without bound is refused as a // transport failure, whether the size is announced or streamed. func TestOversizedResponseIsTransport(t *testing.T) { diff --git a/languages/golang/encrypt/memory_test.go b/languages/golang/encrypt/memory_test.go index 6be1561b5..2ab6becd4 100644 --- a/languages/golang/encrypt/memory_test.go +++ b/languages/golang/encrypt/memory_test.go @@ -202,7 +202,7 @@ func TestRequireLockedMemoryFailsTheCallThatCannotGrow(t *testing.T) { ctx := context.Background() c := strictClient(t) refusing := guest.RefuseGrowth(c.inst.mem, errors.New("refused for the test")) - stage := func(inst *instance) ([]byte, error) { return nil, stageLarge(ctx, inst) } + stage := func(ctx context.Context, inst *instance) ([]byte, error) { return nil, stageLarge(ctx, inst) } _, err := c.call(ctx, stage) if !errors.Is(err, ErrMemoryLock) || !strings.Contains(err.Error(), "growth refused") || !strings.Contains(err.Error(), refusing.Reason().Error()) { t.Fatalf("call needing a refused growth: %v; want ErrMemoryLock naming the refusal", err) @@ -247,7 +247,7 @@ func TestRequireLockedMemoryClosesTheClientOnARefusedInternalGrowth(t *testing.T t.Fatal(err) } var refusing *guest.Refusing - _, err = c.call(ctx, func(inst *instance) ([]byte, error) { + _, err = c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { staged, err := inst.exports.AllocWrite(ctx, inst.module, encoded) if err != nil { return nil, err @@ -268,7 +268,7 @@ func TestRequireLockedMemoryClosesTheClientOnARefusedInternalGrowth(t *testing.T if !c.inst.mem.IsFreed() { t.Fatal("the closed client's memory was not wiped and freed") } - if _, err := c.call(ctx, func(*instance) ([]byte, error) { return nil, nil }); !errors.Is(err, ErrState) { + if _, err := c.call(ctx, func(context.Context, *instance) ([]byte, error) { return nil, nil }); !errors.Is(err, ErrState) { t.Fatalf("a call after the abort: %v, want ErrState", err) } if !c.MemoryLocked() { diff --git a/languages/golang/encrypt/query.go b/languages/golang/encrypt/query.go index b98e76cb5..d4953b4cf 100644 --- a/languages/golang/encrypt/query.go +++ b/languages/golang/encrypt/query.go @@ -131,7 +131,7 @@ func (cph *Cipher) QueryTerm(ctx context.Context, plan *record.Plan, field, oper if err != nil { return QueryTerm{}, err } - out, err := cph.client.call(ctx, func(inst *instance) ([]byte, error) { + out, err := cph.client.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.queryTerm, buf(encodedInput), buf(encodedPlan), buf([]byte(field)), buf([]byte(operator)), buf(opts)) }) if err != nil { @@ -244,7 +244,7 @@ func (cph *Cipher) Verify(ctx context.Context, plan *record.Plan, field, operato if err != nil { return nil, fmt.Errorf("%w: %v", ErrEncoding, err) } - out, err := cph.client.call(ctx, func(inst *instance) ([]byte, error) { + out, err := cph.client.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.verify, buf(encodedInput), buf(encodedValues), buf(encodedPlan), buf([]byte(field)), buf([]byte(operator))) }) if err != nil { @@ -287,7 +287,7 @@ func (cph *Cipher) TargetLimit(ctx context.Context, plan *record.Plan, field, op if err != nil { return Limit{}, fmt.Errorf("%w: %v", ErrEncoding, err) } - out, err := cph.client.call(ctx, func(inst *instance) ([]byte, error) { + out, err := cph.client.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.limit, buf(term), buf(encodedStats), buf(encodedPlan), buf([]byte(field)), buf([]byte(operator)), scalar(want)) }) if err != nil { diff --git a/languages/golang/encrypt/records.go b/languages/golang/encrypt/records.go index 7a4748bbc..b2d0ea745 100644 --- a/languages/golang/encrypt/records.go +++ b/languages/golang/encrypt/records.go @@ -65,7 +65,7 @@ func (cph *Cipher) Seal(ctx context.Context, plan *record.Plan, rows []record.So if err != nil { return nil, err } - out, err := cph.client.call(ctx, func(inst *instance) ([]byte, error) { + out, err := cph.client.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.encryptRecord, buf(encodedSource), buf(encodedPlan), buf(opts)) }) if errors.Is(err, ErrTerm) { @@ -154,7 +154,7 @@ func (cph *Cipher) derive(ctx context.Context, plan *record.Plan, field string, if err != nil { return nil, err } - return cph.client.call(ctx, func(inst *instance) ([]byte, error) { + return cph.client.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.term, buf(encodedValue), buf(encodedContext), scalar(uint64(code)), buf(opts)) }) } @@ -359,9 +359,9 @@ var errNoContext = errors.New("encrypt: the record has no value for its context var errNoEQL = errors.New("encrypt: the record has no EQL value for a field that names an EQL type") // open decrypts records under the keyset the selector names, and, when -// context is not "", refuses a record whose stored context field names +// expected is not "", refuses a record whose stored context field names // another context before any key is retrieved. -func (c *Client) open(ctx context.Context, sel KeysetSelector, context string, p *record.Plan, records []record.Sealed) ([]record.Source, error) { +func (c *Client) open(ctx context.Context, sel KeysetSelector, expected string, p *record.Plan, records []record.Sealed) ([]record.Source, error) { trees := make([]any, len(records)) for i, rec := range records { tree := make(map[string]any, len(p.Fields)+1) @@ -404,11 +404,11 @@ func (c *Client) open(ctx context.Context, sel KeysetSelector, context string, p if err != nil { return nil, fmt.Errorf("%w: %v", ErrEncoding, err) } - opts, err := vcffi.Marshal(options(sel, context)) + opts, err := vcffi.Marshal(options(sel, expected)) if err != nil { return nil, err } - out, err := c.call(ctx, func(inst *instance) ([]byte, error) { + out, err := c.call(ctx, func(ctx context.Context, inst *instance) ([]byte, error) { return inst.call(ctx, inst.decryptRecord, buf(encodedTree), buf(encodedPlan), buf(opts)) }) if err != nil { diff --git a/languages/golang/encrypt/transport.go b/languages/golang/encrypt/transport.go index 85e349555..2e1005b44 100644 --- a/languages/golang/encrypt/transport.go +++ b/languages/golang/encrypt/transport.go @@ -14,6 +14,8 @@ import ( "github.com/tetratelabs/wazero" "github.com/tetratelabs/wazero/api" + + "github.com/cipherstash/stack/languages/golang/internal/guest" ) // transportModule is the name of the guest's one import module. Its two @@ -117,7 +119,10 @@ func (t *transport) perform(ctx context.Context, mem api.Memory, // included, and a wipe racing that send would put a truncated or // zeroed request on the wire. reqBody := newRequestBody(body) - req, err := http.NewRequestWithContext(ctx, string(method), string(url), reqBody) + // The request runs under the caller's context, not the call's: the + // caller's cancellation ends the request, and the guest sees a + // transport failure it returns through its ordinary error path. + req, err := http.NewRequestWithContext(guest.Caller(ctx), string(method), string(url), reqBody) if err != nil { _ = reqBody.Close() return transportFailed, nil, []byte(err.Error()) @@ -212,7 +217,9 @@ func (b *requestBody) Close() error { // tokenGet is token_get: hand the guest the current bearer token. func (t *transport) tokenGet(ctx context.Context, m api.Module, tokenPtrOut, tokenLenOut uint32) int32 { - token, err := t.token.Token(ctx) + // The caller's context, as for a request: a token fetch or refresh the + // caller gave up on ends, and the client stays open. + token, err := t.token.Token(guest.Caller(ctx)) if err != nil { t.tokenErr = err return hostFailed diff --git a/languages/golang/internal/guest/errors.go b/languages/golang/internal/guest/errors.go index 347ec4609..aec29155f 100644 --- a/languages/golang/internal/guest/errors.go +++ b/languages/golang/internal/guest/errors.go @@ -19,7 +19,7 @@ var ( // it. ErrEncoding = errors.New("cipherstash: malformed input") // ErrState is a call on something that has been closed: a client, a - // store, or an instance an interrupted call took down. + // store, or an instance a trap or a stuck call took down. ErrState = errors.New("cipherstash: closed") // ErrInternal is a guest panic or any other unexpected guest failure. ErrInternal = errors.New("cipherstash: internal guest failure") diff --git a/languages/golang/internal/guest/lock.go b/languages/golang/internal/guest/lock.go new file mode 100644 index 000000000..5452837c0 --- /dev/null +++ b/languages/golang/internal/guest/lock.go @@ -0,0 +1,117 @@ +package guest + +import ( + "context" + "sync" + "time" +) + +// What a caller's context does to a guest call. A guest instance is shared: +// an encrypt.Client or an auth profile serves every request in a process, +// so one caller's cancellation must end that caller's call and nothing +// else. Three pieces make that so, and every guest package uses all three. +// +// - The wait for the instance ends with the caller's context ([Lock]). A +// request whose deadline passes in the queue leaves with the context's +// error and never reaches the guest. +// - The guest runs under a context the caller cannot end ([Detach]). The +// runtime closes a module whose call context ends (the runtimes are +// built with WithCloseOnContextDone), and a closed module is a client +// gone for everyone. The caller's context still reaches the host +// functions ([Caller]), so a ZeroKMS request or a token fetch in flight +// is cancelled, fails as a transport failure, and the guest unwinds +// through its ordinary error path with its keys intact. +// - A guest that has not returned [StuckAfter] after its caller's context +// ended is closed: the call context Detach returns ends then. That is +// the backstop for a guest or host function that ignores cancellation, +// which is a bug, never a closed browser tab. + +// StuckAfter is how long a guest call may go on after its caller's context +// ends before the guest is treated as stuck and closed. Once the network +// wait is cancelled, what remains of a call is the guest's own work, which +// takes milliseconds; the margin is wide so that a large batch cancelled +// mid-way still finishes and leaves the instance open. +const StuckAfter = 30 * time.Second + +// Lock is a guest instance's call lock: one call at a time, as a wasm +// instance is single-threaded, with a wait that ends with the caller's +// context. The zero value is unlocked. +type Lock struct { + once sync.Once + ch chan struct{} +} + +func (l *Lock) slot() chan struct{} { + l.once.Do(func() { l.ch = make(chan struct{}, 1) }) + return l.ch +} + +// Acquire takes the lock, or returns ctx's error if ctx ends first. A +// context that has already ended is refused even when the lock is free, +// so a call that would start with nothing to wait for never starts. +func (l *Lock) Acquire(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } + select { + case l.slot() <- struct{}{}: + case <-ctx.Done(): + return ctx.Err() + } + // Both cases can be ready at once, and select picks either: a context + // that ended while the lock was free must not run the call. + if err := ctx.Err(); err != nil { + l.Unlock() + return err + } + return nil +} + +// Lock takes the lock with no way out, for Close, which must not be +// skippable. +func (l *Lock) Lock() { l.slot() <- struct{}{} } + +// Unlock releases the lock. +func (l *Lock) Unlock() { <-l.slot() } + +// callerKey carries the caller's context inside the one Detach returns. +type callerKey struct{} + +// Detach returns the context a guest call runs under: ctx's values, but not +// its cancellation, so the runtime does not close the module when the +// caller gives up. It ends stuck after ctx does, if the call has not +// finished by then, and the runtime then closes the module. done must be +// called when the call returns; it stops the backstop. +func Detach(ctx context.Context, stuck time.Duration) (run context.Context, done func()) { + run, cancel := context.WithCancel(context.WithValue(context.WithoutCancel(ctx), callerKey{}, ctx)) + var mu sync.Mutex + var timer *time.Timer + finished := false + stop := context.AfterFunc(ctx, func() { + mu.Lock() + defer mu.Unlock() + if !finished { + timer = time.AfterFunc(stuck, cancel) + } + }) + return run, func() { + stop() + mu.Lock() + finished = true + if timer != nil { + timer.Stop() + } + mu.Unlock() + cancel() + } +} + +// Caller is the caller's context inside a host function: what a request to +// the network or a token source must honour, so the caller's cancellation +// reaches it. Outside a call Detach began, it is ctx itself. +func Caller(ctx context.Context) context.Context { + if caller, ok := ctx.Value(callerKey{}).(context.Context); ok { + return caller + } + return ctx +} diff --git a/languages/golang/internal/guest/lock_test.go b/languages/golang/internal/guest/lock_test.go new file mode 100644 index 000000000..d4ab83ec7 --- /dev/null +++ b/languages/golang/internal/guest/lock_test.go @@ -0,0 +1,82 @@ +package guest + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestLockRefusesAnEndedContextEvenWhenFree(t *testing.T) { + var l Lock + ctx, cancel := context.WithCancel(context.Background()) + cancel() + for range 100 { + if err := l.Acquire(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("Acquire with an ended context: %v, want context.Canceled", err) + } + } + // And the refusals left it free. + if err := l.Acquire(context.Background()); err != nil { + t.Fatal(err) + } + l.Unlock() +} + +func TestLockWaitEndsWithTheContext(t *testing.T) { + var l Lock + l.Lock() + defer l.Unlock() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + if err := l.Acquire(ctx); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Acquire on a held lock: %v, want the deadline", err) + } +} + +type valueKey struct{} + +func TestDetachKeepsValuesNotCancellation(t *testing.T) { + caller, cancel := context.WithCancel(context.WithValue(context.Background(), valueKey{}, "v")) + run, done := Detach(caller, time.Hour) + defer done() + cancel() + if run.Err() != nil { + t.Fatal("the caller's cancellation reached the call's context") + } + if run.Value(valueKey{}) != "v" { + t.Error("the caller's values did not reach the call's context") + } + if Caller(run) != caller { + t.Error("Caller does not give back the caller's context") + } + if Caller(caller) != caller { + t.Error("Caller outside a detached call is not the context itself") + } +} + +func TestDetachEndsStuckAfterTheCaller(t *testing.T) { + caller, cancel := context.WithCancel(context.Background()) + run, done := Detach(caller, 10*time.Millisecond) + defer done() + cancel() + select { + case <-run.Done(): + case <-time.After(5 * time.Second): + t.Fatal("the call's context did not end after the caller's") + } +} + +// A call that finished before its caller's context ended keeps nothing +// running: the context the call ran under has ended by done, not by a timer. +func TestDetachDoneEndsTheCall(t *testing.T) { + caller, cancel := context.WithCancel(context.Background()) + defer cancel() + run, done := Detach(caller, time.Hour) + done() + select { + case <-run.Done(): + default: + t.Fatal("done left the call's context running") + } +}