From 92f78d369a02cc0ee0909f78e4d9e471076c66d8 Mon Sep 17 00:00:00 2001 From: Joakim Olsson Date: Wed, 16 Sep 2026 07:03:54 +0200 Subject: [PATCH 1/2] fix: keep existing privileges when User.Added is processed late Setup() registers one transient consumer per routing key, and each mints its own queue drained by its own goroutine, so User.Added and Privilege.Added for the same company can be processed in either order. Process(*UserAdded) replaced the company entry with empty privileges, so a Privilege.Added handled first lost its privilege until the next Fetch(), which only runs at service start. Create the entry only when it is missing instead, matching authz-service's own aggregate and read view, which both leave an existing user entry alone. This also makes a User.Added redelivery harmless. Seen as authz-service #826 acceptance-test failures: a new company's Admin grant vanished, so company-service's CreateCompany callback waited out its 30s timeout and the company page stayed empty. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01XMvdB7bcwn1CrQKM4dCshM --- CLAUDE.md | 2 +- client.go | 15 ++++++++++----- client_test.go | 45 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 56 insertions(+), 6 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 79cd1d8..3fb784e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -43,4 +43,4 @@ The `CompanyPrivileges` struct contains permission flags: ### Event Handling -Registers per-replica (transient) go-messaging-amqp consumers for privilege update events from the authz-service (`Setup()`), keeping the local privilege cache up-to-date. Don't combine `Setup()` with go-messaging-amqp's `WithReconnect`: a reconnect declares new per-replica queues, so revocations published during the outage are lost unless `Fetch()` runs again. Services exit on connection loss (`CloseListener`) and re-fetch on start. +Registers per-replica (transient) go-messaging-amqp consumers for privilege update events from the authz-service (`Setup()`), keeping the local privilege cache up-to-date. Each routing key gets its own queue, so events for one company arrive in any order — `Process` must never discard state it didn't itself record. `User.Added` therefore creates the company entry only when it is missing: a `Privilege.Added` processed ahead of it would otherwise be overwritten, and nothing re-reads the privilege until the next `Fetch()`. The converse doesn't hold yet — `setPrivileges` still creates an entry for a company it never saw a `User.Added` for, so a `Privilege.*` delivered after `User.Removed` resurrects an all-false membership. Don't combine `Setup()` with go-messaging-amqp's `WithReconnect`: a reconnect declares new per-replica queues, so revocations published during the outage are lost unless `Fetch()` runs again. Services exit on connection loss (`CloseListener`) and re-fetch on start. diff --git a/client.go b/client.go index 03dddf1..a506cec 100644 --- a/client.go +++ b/client.go @@ -123,12 +123,17 @@ func (h *PrivilegeHandler) Process(msg any) error { switch ev := msg.(type) { case *UserAdded: - if priv, exists := h.privileges[ev.Email]; exists { + // Keep the privileges already recorded for the company. Each routing key has + // its own transient queue, so a Privilege.Added published after this event can + // be processed before it; overwriting the entry here would drop that privilege + // until the next Fetch, which only runs at start. + priv, exists := h.privileges[ev.Email] + if !exists { + priv = map[string]*CompanyPrivileges{} + h.privileges[ev.Email] = priv + } + if _, exists := priv[ev.CompanyID]; !exists { priv[ev.CompanyID] = &CompanyPrivileges{} - } else { - h.privileges[ev.Email] = map[string]*CompanyPrivileges{ - ev.CompanyID: {}, - } } return nil case *UserRemoved: diff --git a/client_test.go b/client_test.go index ac69459..eaea615 100644 --- a/client_test.go +++ b/client_test.go @@ -12,6 +12,7 @@ import ( goamqp "codeberg.org/messaging/go-messaging-amqp" spec "codeberg.org/messaging/messaging" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestPrivilegeHandler_Process_InvalidType(t *testing.T) { @@ -91,6 +92,50 @@ func TestPrivilegeHandler_Process_UserAdded_And_UserRemoved(t *testing.T) { assert.Empty(t, companies) } +func TestPrivilegeHandler_Process_UserAdded_Keeps_Existing_Privileges(t *testing.T) { + // Each routing key has its own transient queue, so Privilege.Added can be + // processed before the User.Added published before it. The privilege must + // survive either order: nothing re-reads it until the next Fetch at start. + userAdded := &UserAdded{Email: "jim@example.org", CompanyID: "abc-123"} + privilegeAdded := &PrivilegeAdded{Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin} + for name, order := range map[string][]any{ + "user added first": {userAdded, privilegeAdded}, + "privilege added first": {privilegeAdded, userAdded}, + } { + t.Run(name, func(t *testing.T) { + handler := New(WithBaseURL("base")) + for _, event := range order { + require.NoError(t, handler.Process(event)) + } + + assert.True(t, handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool { + return privileges.Admin + })) + }) + } +} + +func TestPrivilegeHandler_Process_UserAdded_After_UserRemoved_Starts_Empty(t *testing.T) { + handler := New(WithBaseURL("base")) + + assert.NoError(t, handler.Process(&UserAdded{Email: "jim@example.org", CompanyID: "abc-123"})) + assert.NoError(t, handler.Process(&PrivilegeAdded{ + Email: "jim@example.org", + CompanyID: "abc-123", + Privilege: PrivilegeAdmin, + })) + assert.NoError(t, handler.Process(&UserRemoved{Email: "jim@example.org", CompanyID: "abc-123"})) + assert.NoError(t, handler.Process(&UserAdded{Email: "jim@example.org", CompanyID: "abc-123"})) + + // Membership is back, without the privileges the removal took away. + assert.Equal(t, []string{"abc-123"}, handler.CompaniesByUser("jim@example.org", func(CompanyPrivileges) bool { + return true + })) + assert.False(t, handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool { + return privileges.Admin + })) +} + func TestPrivilegeHandler_GetCompanies_Email_Not_Found(t *testing.T) { handler := New(WithBaseURL("base")) -- 2.54.0 From 55e16832c197e690e4b3ceb96a78f7a3b1b8652f Mon Sep 17 00:00:00 2001 From: Joakim Olsson Date: Wed, 16 Sep 2026 07:16:09 +0200 Subject: [PATCH 2/2] chore(deps): bump go-messaging-amqp to v0.0.5 and amqp091-go to v1.15.0 govulncheck reports GO-2026-6372 against amqp091-go v1.12.0, pulled in indirectly: a broker-controlled oversized payload can exhaust memory. Fixed in v1.13.0; take v1.15.0. The advisory fails CI on main too, and this is the first build since it was published. go-messaging-amqp goes to v0.0.5 at the same time, which is what every migrated service already runs. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01XMvdB7bcwn1CrQKM4dCshM --- go.mod | 4 ++-- go.sum | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/go.mod b/go.mod index ddfc797..4c24270 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module gitea.unbound.se/shiny/authz_client go 1.26.2 require ( - codeberg.org/messaging/go-messaging-amqp v0.0.4 + codeberg.org/messaging/go-messaging-amqp v0.0.5 codeberg.org/messaging/messaging v0.0.5 github.com/stretchr/testify v1.12.1 ) @@ -19,7 +19,7 @@ require ( github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.66.1 // indirect github.com/prometheus/procfs v0.16.1 // indirect - github.com/rabbitmq/amqp091-go v1.12.0 // indirect + github.com/rabbitmq/amqp091-go v1.15.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/otel v1.44.0 // indirect go.opentelemetry.io/otel/metric v1.44.0 // indirect diff --git a/go.sum b/go.sum index 6ae674b..0ce088c 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -codeberg.org/messaging/go-messaging-amqp v0.0.4 h1:MwY/kU1lBdCL7ywKAg2k6aHuJf8f78MRZQSuyrNFvQ8= -codeberg.org/messaging/go-messaging-amqp v0.0.4/go.mod h1:6bSCIkKH0V/oI7vaIq8ttINKYwov9c8ckZrwjfvp7kc= +codeberg.org/messaging/go-messaging-amqp v0.0.5 h1:zzcJ+FVWgBPKyTkq9RXF11yvnhs8IN5BALBC+mps1zA= +codeberg.org/messaging/go-messaging-amqp v0.0.5/go.mod h1:6bSCIkKH0V/oI7vaIq8ttINKYwov9c8ckZrwjfvp7kc= codeberg.org/messaging/messaging v0.0.5 h1:/ueH90F4RNPUeeJIwdG9oVhDW7+XjCzwOYq8xc0kfQk= codeberg.org/messaging/messaging v0.0.5/go.mod h1:xyWLUcfaVzcN6GWk9uaQsxPVUC2Vm2S89mYZGdiWTWg= codeberg.org/messaging/messaging/tck v0.0.3 h1:a4Nr7ytFqEJloTOyVeAtMNgO9Fig1ZUirjW4FVlK0lc= @@ -39,8 +39,8 @@ github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9Z github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg= github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is= -github.com/rabbitmq/amqp091-go v1.12.0 h1:V0v14Iqfs+MwHWihJt/nGS5Ulu0vw572b2Co3mwunkI= -github.com/rabbitmq/amqp091-go v1.12.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= +github.com/rabbitmq/amqp091-go v1.15.0 h1:LEQL4/yp48/Wigt6A6XOu18RQRo8ZHtB5I/KZJn+gkw= +github.com/rabbitmq/amqp091-go v1.15.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= -- 2.54.0