Vault 2256: fix lease count quotas causing panics on dr secondaries (#11742)

* lift relevant changes from ent to oss

* fix silent error bug in quotas

vault/core.go

@@ -2724,7 +2724,7 @@ func (c *Core) setupQuotas(ctx context.Context, isPerfStandby bool) error {
 		return nil
 	}
 
-	return c.quotaManager.Setup(ctx, c.systemBarrierView, isPerfStandby)
+	return c.quotaManager.Setup(ctx, c.systemBarrierView, isPerfStandby, c.IsDRSecondary())
 }
 
 // ApplyRateLimitQuota checks the request against all the applicable quota rules.

vault/quotas/quotas.go

@@ -771,6 +771,11 @@ func (m *Manager) Invalidate(key string) {
 		qType := splitKeys[0]
 		name := splitKeys[1]
 
+		if qType == TypeLeaseCount.String() && m.isDRSecondary {
+			// lease count invalidation not supported on DR Secondary
+			return
+		}
+
 		// Read quota rule from storage
 		quota, err := Load(m.ctx, m.storage, qType, name)
 		if err != nil {
@@ -844,13 +849,14 @@ func Load(ctx context.Context, storage logical.Storage, qType, name string) (Quo
 
 // Setup loads the quota configuration and all the quota rules into the
 // quota manager.
-func (m *Manager) Setup(ctx context.Context, storage logical.Storage, isPerfStandby bool) error {
+func (m *Manager) Setup(ctx context.Context, storage logical.Storage, isPerfStandby, isDRSecondary bool) error {
 	m.lock.Lock()
 	defer m.lock.Unlock()
 
 	m.storage = storage
 	m.ctx = ctx
 	m.isPerfStandby = isPerfStandby
+	m.isDRSecondary = isDRSecondary
 
 	// Load the quota configuration from storage and load it into the quota
 	// manager.
@@ -887,15 +893,25 @@ func (m *Manager) Setup(ctx context.Context, storage logical.Storage, isPerfStan
 		return err
 	}
 
-	// Load the quota rules for all supported types from storage and load it in
-	// the quota manager.
 	for _, qType := range quotaTypes() {
-		names, err := logical.CollectKeys(ctx, logical.NewStorageView(storage, StoragePrefix+qType+"/"))
-		if err != nil {
+		m.setupQuotaType(ctx, storage, qType)
+	}
+
 	return nil
 }
+
+func (m *Manager) setupQuotaType(ctx context.Context, storage logical.Storage, quotaType string) error {
+	if quotaType == TypeLeaseCount.String() && m.isDRSecondary {
+		m.logger.Trace("lease count quotas are not processed on DR Secondaries")
+		return nil
+	}
+
+	names, err := logical.CollectKeys(ctx, logical.NewStorageView(storage, StoragePrefix+quotaType+"/"))
+	if err != nil {
+		return err
+	}
 	for _, name := range names {
-			quota, err := Load(ctx, m.storage, qType, name)
+		quota, err := Load(ctx, m.storage, quotaType, name)
 		if err != nil {
 			return err
 		}
@@ -904,12 +920,11 @@ func (m *Manager) Setup(ctx context.Context, storage logical.Storage, isPerfStan
 			continue
 		}
 
-			err = m.setQuotaLocked(ctx, qType, quota, true)
+		err = m.setQuotaLocked(ctx, quotaType, quota, true)
 		if err != nil {
 			return err
 		}
 	}
-	}
 
 	return nil
 }

vault/quotas/quotas_util.go

@@ -31,6 +31,7 @@ func (m *Manager) inLeasePathCache(path string) bool {
 
 type entManager struct {
 	isPerfStandby bool
+	isDRSecondary bool
 }
 
 func (*entManager) Reset() error {