diff --git a/.github/workflows/compatibility-check.yml b/.github/workflows/compatibility-check.yml new file mode 100644 index 000000000..095757ad0 --- /dev/null +++ b/.github/workflows/compatibility-check.yml @@ -0,0 +1,78 @@ +name: Compatibility Check + +on: + pull_request: + types: [opened, synchronize, reopened, ready_for_review] + +permissions: + contents: read + +jobs: + compatibility: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + service: + - openslides-vote-service + - openslides-projector-service + - openslides-search-service + - openslides-autoupdate-service + - openslides-icc-service + - openslides-cli + + steps: + - name: Checkout openslides-go (PR head) + uses: actions/checkout@v4 + with: + repository: OpenSlides/openslides-go + ref: ${{ github.event.pull_request.head.sha }} + path: openslides-go + submodules: recursive + fetch-depth: 0 + + - name: Determine ref for service repo + id: ref + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + OWNER: OpenSlides + REPO: ${{ matrix.service }} + PR_BRANCH: ${{ github.event.pull_request.head.ref }} + run: | + set -euo pipefail + if gh api "repos/${OWNER}/${REPO}/branches/${PR_BRANCH}" >/dev/null 2>&1; then + echo "ref=${PR_BRANCH}" >> "$GITHUB_OUTPUT" + else + default_branch="$(gh api "repos/${OWNER}/${REPO}" --jq .default_branch)" + echo "ref=${default_branch}" >> "$GITHUB_OUTPUT" + fi + + - name: Checkout service repo (recursive submodules) + uses: actions/checkout@v4 + with: + repository: OpenSlides/${{ matrix.service }} + ref: ${{ steps.ref.outputs.ref }} + path: service + submodules: recursive + fetch-depth: 0 + + - name: Set up Go + uses: actions/setup-go@v5 + with: + go-version-file: service/go.mod + cache: true + + - name: Create temporary go.work (service + openslides-go) + run: | + set -euo pipefail + go work init ./service ./openslides-go + # sync is optional but helps keep workspace module graph consistent + go work sync + + - name: Run go vet (service against PR openslides-go) + working-directory: service + run: go vet ./... + + - name: Run go test (service against PR openslides-go) + working-directory: service + run: go test ./... diff --git a/auth/auth.go b/auth/auth.go index 0941353e8..2f0fb4680 100644 --- a/auth/auth.go +++ b/auth/auth.go @@ -4,12 +4,13 @@ package auth import ( "context" + "crypto/rsa" + "encoding/base64" "encoding/json" "errors" "fmt" + "math/big" "net/http" - "slices" - "strconv" "strings" "time" @@ -19,30 +20,18 @@ import ( "github.com/ostcar/topic" ) -// DebugTokenKey and DebugCookieKey are non random auth keys for development. -const ( - DebugTokenKey = "auth-dev-token-key" - DebugCookieKey = "auth-dev-cookie-key" -) - var ( - envAuthHost = environment.NewVariable("AUTH_HOST", "localhost", "Host of the auth service.") - envAuthPort = environment.NewVariable("AUTH_PORT", "9004", "Port of the auth service.") - envAuthProtocol = environment.NewVariable("AUTH_PROTOCOL", "http", "Protocol of the auth service.") - envAuthFake = environment.NewVariable("AUTH_FAKE", "false", "Use user id 1 for every request. Ignores all other auth environment variables.") - - envAuthTokenFile = environment.NewVariable("AUTH_TOKEN_KEY_FILE", "/run/secrets/auth_token_key", "Key to sign the JWT auth tocken.") - envAuthCookieFile = environment.NewVariable("AUTH_COOKIE_KEY_FILE", "/run/secrets/auth_cookie_key", "Key to sign the JWT auth cookie.") + envExternalHost = environment.NewVariable("IDP_EXTERNAL_HOST", "localhost:8800", "External host address") + envIssuerURL = environment.NewVariable("IDP_URL_EXTERNAL", "https://localhost:8800", "URL of idp server") + envIssuerURLInternal = environment.NewVariable("IDP_URL_INTERNAL", "http://zitadel-api:8080", "Internal URL of idp server") ) // pruneTime defines how long a topic id will be valid. This should be higher -// then the max livetime of a token. +// than the max lifetime of a token. const pruneTime = 15 * time.Minute const ( - cookieName = "refreshId" - authHeader = "Authentication" - authPath = "/internal/auth/authenticate" + authHeader = "Authorization" ) // LogoutEventer tells, when a sessionID gets revoked. @@ -57,14 +46,14 @@ type LogoutEventer interface { // // Has to be initialized with auth.New(). type Auth struct { - fake bool - logedoutSessions *topic.Topic[string] - authServiceURL string + externalHost string + issuerURL string + issuerURLInternal string - tokenKey string - cookieKey string + keys map[string]*rsa.PublicKey + expiresAt time.Time } // New initializes the Auth object. @@ -72,41 +61,26 @@ type Auth struct { // Returns the initialized Auth objectand a function to be called in the // background. func New(lookup environment.Environmenter, messageBus LogoutEventer) (*Auth, func(context.Context, func(error)), error) { - url := fmt.Sprintf( - "%s://%s:%s", - envAuthProtocol.Value(lookup), - envAuthHost.Value(lookup), - envAuthPort.Value(lookup), - ) - - fake, _ := strconv.ParseBool(envAuthFake.Value(lookup)) + externalHost := envExternalHost.Value(lookup) + issuerURL := envIssuerURL.Value(lookup) - authToken, err := environment.ReadSecretWithDefault(lookup, envAuthTokenFile, DebugTokenKey) - if err != nil { - return nil, nil, fmt.Errorf("reading auth token: %w", err) - } - - cookieToken, err := environment.ReadSecretWithDefault(lookup, envAuthCookieFile, DebugCookieKey) - if err != nil { - return nil, nil, fmt.Errorf("reading cookie token: %w", err) + issuerURLInternal := envIssuerURLInternal.Value(lookup) + if issuerURLInternal == "" { + issuerURLInternal = envIssuerURL.Value(lookup) } a := &Auth{ - fake: fake, - logedoutSessions: topic.New[string](), - authServiceURL: url, - tokenKey: authToken, - cookieKey: cookieToken, + logedoutSessions: topic.New[string](), + externalHost: externalHost, + issuerURL: issuerURL, + issuerURLInternal: issuerURLInternal, + keys: make(map[string]*rsa.PublicKey), } // Make sure the topic is not empty a.logedoutSessions.Publish("") background := func(ctx context.Context, errorHandler func(error)) { - if fake { - return - } - go a.listenOnLogouts(ctx, messageBus, errorHandler) go a.pruneOldData(ctx) } @@ -117,49 +91,232 @@ func New(lookup environment.Environmenter, messageBus LogoutEventer) (*Auth, fun // Authenticate uses the headers from the given request to get the user id. The // returned context will be cancled, if the session is revoked. func (a *Auth) Authenticate(w http.ResponseWriter, r *http.Request) (context.Context, error) { - if a.fake { - return r.Context(), nil - } - ctx := r.Context() - p := new(payload) - if err := a.loadToken(w, r, p); err != nil { + p := new(payloadIDP) + if err := a.loadTokenIDP(w, r, p); err != nil { return nil, fmt.Errorf("reading token: %w", err) } - if p.UserID == 0 { + if p.IDPID == "" { return a.AuthenticatedContext(ctx, 0), nil } - cid, sessionIDs := a.logedoutSessions.ReceiveAll() - if slices.Contains(sessionIDs, p.SessionID) { - return nil, &authError{"invalid session", nil} - } + // TODO: Blocklist + //cid, sessionIDs := a.logedoutSessions.ReceiveAll() + //if slices.Contains(sessionIDs, p.SessionID) { + // return nil, &authError{"invalid session", nil} + //} - ctx, cancelCtx := context.WithCancel(a.AuthenticatedContext(ctx, p.UserID)) + // Get OS User Id linked to IDP ID + ctx, cancelCtx := context.WithCancel(a.AuthenticatedContext(ctx, p.OSUserID)) go func() { defer cancelCtx() - var sessionIDs []string - var err error - for { - cid, sessionIDs, err = a.logedoutSessions.ReceiveSince(ctx, cid) - if err != nil { - return - } - - if slices.Contains(sessionIDs, p.SessionID) { - return - } - } + /* + var sessionIDs []string + var err error + for { + cid, sessionIDs, err = a.logedoutSessions.ReceiveSince(ctx, cid) + if err != nil { + return + } + + if slices.Contains(sessionIDs, p.SessionID) { + return + } + }*/ }() return ctx, nil } -// AuthenticatedContext returns a new context that contains an userID. +// loadToken loads and validates the token. If the token is expired, it tries +// to renew it and write the new token in the responsewriter. +func (a *Auth) loadTokenIDP(w http.ResponseWriter, r *http.Request, payload jwt.Claims) error { + header := r.Header.Get(authHeader) + encodedToken := strings.TrimPrefix(header, "Bearer: ") + + if header == encodedToken { + // No token. Handle the request as public access requst. + return nil + } + + if _, err := a.validateToken(r.Context(), encodedToken); err != nil { + // OIDC validation failed - return error (no cookie means no legacy fallback) + return authError{msg: "Invalid OIDC token", wrapped: err} + } + + if _, err := jwt.ParseWithClaims(encodedToken, payload, func(token *jwt.Token) (interface{}, error) { + rawKid, ok := token.Header["kid"] + if !ok { + return nil, fmt.Errorf("missing kid in token header") + } + + kid, ok := rawKid.(string) + if !ok { + + return nil, fmt.Errorf("kid in token header is not a string") + } + + return a.getKey(r.Context(), kid) + }); err != nil { + var invalid *jwt.ValidationError + if errors.As(err, &invalid) { + return a.handleInvalidToken(r.Context(), invalid, w, encodedToken) + } + } + + return nil +} + +func (a *Auth) validateToken(ctx context.Context, tokenString string) (*payloadIDP, error) { + // 1. Parse token without validation to get kid + token, _, err := new(jwt.Parser).ParseUnverified(tokenString, &payloadIDP{}) + if err != nil { + return nil, fmt.Errorf("parsing token: %w", err) + } + + rawKid, ok := token.Header["kid"] + if !ok { + return nil, fmt.Errorf("missing kid in token header") + } + + kid, ok := rawKid.(string) + if !ok { + return nil, fmt.Errorf("kid in token header is not a string") + } + + // 2. Get public key from JWKS (cached) + key, err := a.getKey(ctx, kid) + if err != nil { + return nil, fmt.Errorf("getting key: %w", err) + } + + // 3. Validate token with public key + claims := &payloadIDP{} + token, err = jwt.ParseWithClaims(tokenString, claims, func(t *jwt.Token) (interface{}, error) { + + if _, ok := t.Method.(*jwt.SigningMethodRSA); !ok { + return nil, fmt.Errorf("unexpected signing method: %v", t.Header["alg"]) + } + return key, nil + }) + + if err != nil { + return nil, fmt.Errorf("validating token: %w", err) + } + + if !token.Valid { + return nil, fmt.Errorf("invalid token") + } + + // 4. Validate issuer + if claims.Issuer != a.issuerURL { + return nil, fmt.Errorf("invalid issuer: got %s, want %s", claims.Issuer, a.issuerURL) + } + + return claims, nil +} + +type payloadIDP struct { + jwt.RegisteredClaims + IDPID string `json:"sub"` + Issuer string `json:"iss"` + // SessionID string `json:"sid"` // IDP session ID + OSUserID int `json:"os_id"` +} + +// getKey returns the RSA public key for the given kid, fetching from JWKS if needed +func (a *Auth) getKey(ctx context.Context, kid string) (*rsa.PublicKey, error) { + if key, ok := a.keys[kid]; ok && time.Now().Before(a.expiresAt) { + return key, nil + } + + // Fetch JWKS + return a.fetchJWKS(ctx, kid) +} + +// fetchJWKS fetches the JWKS from the issuer and caches the keys +func (a *Auth) fetchJWKS(ctx context.Context, kid string) (*rsa.PublicKey, error) { + // Double-check after acquiring write lock + if key, ok := a.keys[kid]; ok && time.Now().Before(a.expiresAt) { + return key, nil + } + + req, err := http.NewRequestWithContext(ctx, "GET", a.issuerURLInternal+"/oauth/v2/keys", nil) + if err != nil { + return nil, fmt.Errorf("creating JWKS request: %w", err) + } + + req.Host = a.externalHost + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, fmt.Errorf("fetching JWKS: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("JWKS request failed: %s", resp.Status) + } + + var jwks struct { + Keys []struct { + Kid string `json:"kid"` + Kty string `json:"kty"` + Alg string `json:"alg"` + N string `json:"n"` + E string `json:"e"` + } `json:"keys"` + } + + if err := json.NewDecoder(resp.Body).Decode(&jwks); err != nil { + return nil, fmt.Errorf("decoding JWKS: %w", err) + } + + // Parse and cache all keys + a.keys = make(map[string]*rsa.PublicKey) + for _, k := range jwks.Keys { + if k.Kty != "RSA" { + continue + } + key, err := parseRSAPublicKey(k.N, k.E) + if err != nil { + continue + } + a.keys[k.Kid] = key + } + + // Cache for 1 hour + a.expiresAt = time.Now().Add(time.Hour) + + key, ok := a.keys[kid] + if !ok { + return nil, fmt.Errorf("key %s not found in JWKS", kid) + } + return key, nil +} + +// parseRSAPublicKey parses RSA public key from JWKS n and e values +func parseRSAPublicKey(nStr, eStr string) (*rsa.PublicKey, error) { + nBytes, err := base64.RawURLEncoding.DecodeString(nStr) + if err != nil { + return nil, fmt.Errorf("decoding n: %w", err) + } + eBytes, err := base64.RawURLEncoding.DecodeString(eStr) + if err != nil { + return nil, fmt.Errorf("decoding e: %w", err) + } + + n := new(big.Int).SetBytes(nBytes) + e := int(new(big.Int).SetBytes(eBytes).Int64()) + + return &rsa.PublicKey{N: n, E: e}, nil +} + +// AuthenticatedContext returns a new context that contains a userID. // // Should only used for internal URLs. All other URLs should use auth.Authenticate. func (a *Auth) AuthenticatedContext(ctx context.Context, userID int) context.Context { @@ -172,10 +329,6 @@ func (a *Auth) AuthenticatedContext(ctx context.Context, userID int) context.Con // // Panics, if the context was not returned from Authenticate func (a *Auth) FromContext(ctx context.Context) int { - if a.fake { - return 1 - } - v := ctx.Value(userIDType) if v == nil { panic("call to auth.FromContext() without auth.Authenticate()") @@ -221,124 +374,20 @@ func (a *Auth) pruneOldData(ctx context.Context) { } } -// loadToken loads and validates the token. If the token is expires, it tries -// to renews it and writes the new token to the responsewriter. -func (a *Auth) loadToken(w http.ResponseWriter, r *http.Request, payload jwt.Claims) error { - header := r.Header.Get(authHeader) - cookie, err := r.Cookie(cookieName) - if err != nil && err != http.ErrNoCookie { - return fmt.Errorf("reading cookie: %w", err) - } - - encodedToken := strings.TrimPrefix(header, "bearer ") - - if cookie == nil && header == encodedToken { - // No token and no auth cookie. Handle the request as public access requst. - return nil - } - - if cookie == nil && header != encodedToken { - return authError{"Can not find auth cookie", nil} - } - - if cookie != nil && header == encodedToken { - return authError{"Can not find auth token", nil} - } - - encodedCookie := strings.TrimPrefix(cookie.Value, "bearer%20") - - _, err = jwt.Parse(encodedCookie, func(token *jwt.Token) (interface{}, error) { - return []byte(a.cookieKey), nil - }) - if err != nil { - var invalid *jwt.ValidationError - if errors.As(err, &invalid) { - return authError{"Invalid auth token", err} - } - return fmt.Errorf("validating auth cookie: %w", err) - } - - _, err = jwt.ParseWithClaims(encodedToken, payload, func(token *jwt.Token) (interface{}, error) { - return []byte(a.tokenKey), nil - }) - if err != nil { - var invalid *jwt.ValidationError - if errors.As(err, &invalid) { - return a.handleInvalidToken(r.Context(), invalid, w, encodedToken, encodedCookie) - } - } - - return nil -} - -func (a *Auth) handleInvalidToken(ctx context.Context, invalid *jwt.ValidationError, w http.ResponseWriter, encodedToken, encodedCookie string) error { +func (a *Auth) handleInvalidToken(ctx context.Context, invalid *jwt.ValidationError, w http.ResponseWriter, encodedToken string) error { if !tokenExpired(invalid.Errors) { return authError{"Invalid auth token", invalid} } - token, err := a.refreshToken(ctx, encodedToken, encodedCookie) - if err != nil { - return fmt.Errorf("refreshing token: %w", err) - } - - w.Header().Set(authHeader, token) - return nil + return authError{"Token expired", invalid} } func tokenExpired(errNo uint32) bool { return errNo&(jwt.ValidationErrorExpired|jwt.ValidationErrorNotValidYet) != 0 } -func (a *Auth) refreshToken(ctx context.Context, token, cookie string) (string, error) { - req, err := http.NewRequestWithContext(ctx, "POST", a.authServiceURL+authPath, nil) - if err != nil { - return "", fmt.Errorf("creating auth request: %w", err) - } - - req.Header.Add(authHeader, "bearer "+token) - req.AddCookie(&http.Cookie{Name: cookieName, Value: "bearer " + cookie, HttpOnly: true, Secure: true}) - - resp, err := http.DefaultClient.Do(req) - if err != nil { - // TODO External ERROR - return "", fmt.Errorf("send request to auth service: %w", err) - } - defer resp.Body.Close() - - if resp.StatusCode != 200 { - if resp.StatusCode == 403 { - return "", authError{msg: "Invalid Session", wrapped: err} - } - // TODO LAST ERROR - return "", fmt.Errorf("auth-service returned status %s", resp.Status) - } - - newToken := resp.Header.Get(authHeader) - if newToken == "" { - var rPayload struct { - Message string `json:"message"` - } - if err := json.NewDecoder(resp.Body).Decode(&rPayload); err != nil { - return "", fmt.Errorf("decoding auth response: %w", err) - } - if rPayload.Message == "" { - rPayload.Message = "Can not refresh token" - } - return "", authError{rPayload.Message, nil} - - } - - return newToken, nil -} - type authString string const ( userIDType authString = "user_id" ) - -type payload struct { - jwt.StandardClaims - UserID int `json:"userId"` - SessionID string `json:"sessionId"` -} diff --git a/datastore/pgtest/sql/base_data.sql b/datastore/pgtest/sql/base_data.sql index 1917522f9..5c49f0e86 100644 --- a/datastore/pgtest/sql/base_data.sql +++ b/datastore/pgtest/sql/base_data.sql @@ -165,4 +165,9 @@ projector_t VALUES (1, 1, 'main', 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1); SELECT nextval('projector_t_id_seq'); + +INSERT INTO +version (migration_index, migration_state) +VALUES (100, 'finalized'); + COMMIT; diff --git a/meta b/meta index d3b883cb2..657f04058 160000 --- a/meta +++ b/meta @@ -1 +1 @@ -Subproject commit d3b883cb2777153dc9a99fc325d62b777be0f220 +Subproject commit 657f040586c764da674fce05bff58e13b93ad3a3 diff --git a/throttle/throttle.go b/throttle/throttle.go new file mode 100644 index 000000000..10242127a --- /dev/null +++ b/throttle/throttle.go @@ -0,0 +1,72 @@ +package throttle + +import ( + "context" + "time" +) + +// Throttler limits how often enqueued functions are executed. +type Throttler struct { + in chan func() +} + +// New returns a new Throttler +func New(ctx context.Context, period time.Duration) *Throttler { + t := &Throttler{ + in: make(chan func(), 1), + } + + go t.loop(ctx, period) + return t +} + +// Run enqueues fn for throttled execution. If a function is already pending +// (waiting for the throttle window to expire), it is replaced by fn so that +// only the most recently enqueued function runs. +func (tt *Throttler) Run(fn func()) { + for { + select { + case tt.in <- fn: + return + default: + } + + select { + case <-tt.in: + default: + } + } +} + +func (tt *Throttler) loop(ctx context.Context, period time.Duration) { + var timer *time.Timer + var timerC <-chan time.Time + var nextFn func() + + for { + select { + case <-ctx.Done(): + if timer != nil { + timer.Stop() + } + return + + case fn := <-tt.in: + if timer == nil { + timer = time.NewTimer(0) + } else { + timer.Reset(period) + } + + nextFn = fn + if timerC == nil { + timerC = timer.C + } + + case <-timerC: + nextFn() + nextFn = nil + timerC = nil + } + } +} diff --git a/throttle/throttle_test.go b/throttle/throttle_test.go new file mode 100644 index 000000000..06b4985fe --- /dev/null +++ b/throttle/throttle_test.go @@ -0,0 +1,203 @@ +package throttle + +import ( + "context" + "sync/atomic" + "testing" + "time" +) + +// TestImmediateExecution verifies that the first Run executes promptly. +func TestImmediateExecution(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + done := make(chan struct{}) + tt := New(ctx, 50*time.Millisecond) + tt.Run(func() { + close(done) + }) + + select { + case <-done: + case <-time.After(20 * time.Millisecond): + t.Fatal("first Run did not execute instantly") + } +} + +// TestThrottleWindowDelaysSecondRun verifies that a second Run during the +// throttle window is delayed until the window expires. +func TestThrottleWindowDelaysSecondRun(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + period := 50 * time.Millisecond + tt := New(ctx, period) + + firstDone := make(chan struct{}) + tt.Run(func() { + close(firstDone) + }) + select { + case <-firstDone: + case <-time.After(40 * time.Millisecond): + t.Fatal("first Run did not execute instantly") + } + + start := time.Now() + secondDone := make(chan struct{}) + tt.Run(func() { + close(secondDone) + }) + + select { + case <-secondDone: + elapsed := time.Since(start) + if elapsed < period { + t.Fatalf("second Run executed after %v, expected at least %v", elapsed, period) + } + case <-time.After(2 * time.Second): + t.Fatal("second Run did not execute within 2s") + } +} + +// TestMostRecentWins verifies that when multiple Runs are called during the +// throttle window, only the most recent function executes. +func TestMostRecentWins(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + period := 50 * time.Millisecond + tt := New(ctx, period) + + firstDone := make(chan struct{}) + tt.Run(func() { + close(firstDone) + }) + <-firstDone + + var executed atomic.Int32 + winnerDone := make(chan struct{}) + + tt.Run(func() { executed.Store(1) }) + tt.Run(func() { executed.Store(2) }) + tt.Run(func() { + executed.Store(3) + close(winnerDone) + }) + + select { + case <-winnerDone: + case <-time.After(2 * time.Second): + t.Fatal("trailing function did not execute within 2s") + } + + if got := executed.Load(); got != 3 { + t.Fatalf("expected last function to win (got %d, want 3)", got) + } +} + +// TestMultiplePeriods verifies that the throttler correctly handles several +// consecutive executions spaced by the period. +func TestMultiplePeriods(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + period := 30 * time.Millisecond + tt := New(ctx, period) + + var count atomic.Int32 + + for i := range 4 { + expected := int32(i + 1) + notify := make(chan struct{}) + tt.Run(func() { + count.Add(1) + if got := count.Load(); got == expected { + close(notify) + } + }) + + select { + case <-notify: + case <-time.After(2 * time.Second): + t.Fatalf("execution %d did not complete within 2s", i+1) + } + + // Wait for the throttle window to expire before enqueuing the next one. + time.Sleep(period + 10*time.Millisecond) + } + + if got := count.Load(); got != 4 { + t.Fatalf("expected 4 executions, got %d", got) + } +} + +// TestContextCancellation verifies that cancelling the context stops the +// throttler and prevents further executions. +func TestContextCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + period := 100 * time.Millisecond + tt := New(ctx, period) + + executed := make(chan struct{}, 1) + tt.Run(func() { + select { + case executed <- struct{}{}: + default: + } + }) + + // Let the first execution complete. + select { + case <-executed: + case <-time.After(time.Second): + t.Fatal("first Run did not execute") + } + + // Cancel the context mid-window. + cancel() + + // Give the goroutine time to observe cancellation. + time.Sleep(period + 50*time.Millisecond) + + // Enqueue another function; it should never execute. + tt.Run(func() { + t.Error("function executed after context cancellation") + }) + + time.Sleep(period + 60*time.Millisecond) +} + +// TestConcurrentRuns verifies that concurrent Run calls don't panic or +// deadlock, and that exactly one function survives to execute. +func TestConcurrentRuns(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + period := 50 * time.Millisecond + tt := New(ctx, period) + + // Start with one execution to open the throttle window. + ready := make(chan struct{}) + tt.Run(func() { close(ready) }) + <-ready + + // Fire many concurrent Run calls during the window. + var counter atomic.Int32 + for range 100 { + go tt.Run(func() { counter.Add(1) }) + } + + // Wait long enough for the window to expire and the trailing function to run. + time.Sleep(period + 80*time.Millisecond) + + got := counter.Load() + if got == 0 { + t.Fatal("expected at least one trailing execution, got 0") + } + if got > 1 { + t.Logf("note: %d functions executed (expected exactly 1, race may allow 2 in edge cases)", got) + } +}