mirror of
https://github.com/coder/coder.git
synced 2026-09-24 15:04:27 +08:00
fix(enterprise/coderd): propagate license events over Postgres pubsub when NATS is in use (#26536)
Co-authored-by: Mux <mux@coder.com>
This commit is contained in:
@@ -1354,7 +1354,8 @@ func (api *API) runEntitlementsLoop(ctx context.Context) {
|
||||
// the system will eventually recover as replicas timeout
|
||||
// if their heartbeats stop. The best effort just tries to update the
|
||||
// UI faster if it succeeds.
|
||||
_ = api.Pubsub.Publish(PubsubEventLicenses, []byte("going away"))
|
||||
// Postgres pubsub; see PubsubEventLicenses.
|
||||
_ = api.ReplicaSyncPubsub.Publish(PubsubEventLicenses, []byte("going away"))
|
||||
}()
|
||||
for {
|
||||
select {
|
||||
@@ -1364,7 +1365,10 @@ func (api *API) runEntitlementsLoop(ctx context.Context) {
|
||||
// pass
|
||||
}
|
||||
if !subscribed {
|
||||
cancel, err := api.Pubsub.Subscribe(PubsubEventLicenses, func(_ context.Context, _ []byte) {
|
||||
// Postgres pubsub; see PubsubEventLicenses. ReplicaSyncPubsub is
|
||||
// always set in enterprise startup (replicasync.New requires it when
|
||||
// the API is constructed), so it is safe to use directly here.
|
||||
cancel, err := api.ReplicaSyncPubsub.Subscribe(PubsubEventLicenses, func(_ context.Context, _ []byte) {
|
||||
// don't block. If the channel is full, drop the event, as there is a resync
|
||||
// scheduled already.
|
||||
select {
|
||||
|
||||
@@ -33,6 +33,16 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
// PubsubEventLicenses is the pubsub event that tells other replicas to
|
||||
// re-read licenses from the database and recompute entitlements.
|
||||
//
|
||||
// It is published and subscribed on api.ReplicaSyncPubsub (always Postgres),
|
||||
// not api.Pubsub. When the NATS pubsub experiment is enabled, api.Pubsub is
|
||||
// the embedded NATS pubsub whose cluster mesh only forms once a replica is
|
||||
// HA-licensed, so propagating license changes over it is circular: a fresh
|
||||
// replica could not learn about the license that would let it join the mesh.
|
||||
// ReplicaSyncPubsub is available as soon as the DB connection is, independent
|
||||
// of clustering or licensing.
|
||||
PubsubEventLicenses = "licenses"
|
||||
)
|
||||
|
||||
@@ -136,7 +146,8 @@ func (api *API) postLicense(rw http.ResponseWriter, r *http.Request) {
|
||||
})
|
||||
return
|
||||
}
|
||||
err = api.Pubsub.Publish(PubsubEventLicenses, []byte("add"))
|
||||
// Postgres pubsub; see PubsubEventLicenses.
|
||||
err = api.ReplicaSyncPubsub.Publish(PubsubEventLicenses, []byte("add"))
|
||||
if err != nil {
|
||||
api.Logger.Error(context.Background(), "failed to publish license add", slog.Error(err))
|
||||
// don't fail the HTTP request, since we did write it successfully to the database
|
||||
@@ -217,7 +228,8 @@ func (api *API) refreshEntitlements(ctx context.Context) error {
|
||||
if err != nil {
|
||||
return xerrors.Errorf("failed to update entitlements: %w", err)
|
||||
}
|
||||
err = api.Pubsub.Publish(PubsubEventLicenses, []byte("refresh"))
|
||||
// Postgres pubsub; see PubsubEventLicenses.
|
||||
err = api.ReplicaSyncPubsub.Publish(PubsubEventLicenses, []byte("refresh"))
|
||||
if err != nil {
|
||||
api.Logger.Error(ctx, "failed to publish forced entitlement update", slog.Error(err))
|
||||
return xerrors.Errorf("failed to publish forced entitlement update, other replicas might not be updated: %w", err)
|
||||
@@ -331,7 +343,8 @@ func (api *API) deleteLicense(rw http.ResponseWriter, r *http.Request) {
|
||||
})
|
||||
return
|
||||
}
|
||||
err = api.Pubsub.Publish(PubsubEventLicenses, []byte("delete"))
|
||||
// Postgres pubsub; see PubsubEventLicenses.
|
||||
err = api.ReplicaSyncPubsub.Publish(PubsubEventLicenses, []byte("delete"))
|
||||
if err != nil {
|
||||
api.Logger.Error(context.Background(), "failed to publish license delete", slog.Error(err))
|
||||
// don't fail the HTTP request, since we did write it successfully to the database
|
||||
|
||||
Reference in New Issue
Block a user