mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-03 03:56:04 +00:00
ShivamandGitHub
1cd4b7d0bb
## What changed? - WISOTT ## Why? - [this](https://github.com/temporalio/temporal/pull/9386) PR put forward flakes in the task queue stats suite. The main reason for those flakes were only occurring when each subtest was being run in parallel to each other using TestEnv. - The reason why those flakes started to appear was because the getLocked method, in the memory client, returns the last writer's value for a specific dynamic config key. The subtest in the flaky test suite are all writing different values to the same dynamic config key, which is resulting in race conditions during test execution. ## How did you test it? - [ ] built - [ ] run locally and tested manually - [ ] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) ## Potential risks - None --- > [!NOTE] > **Medium Risk** > Changes `MemoryClient` override resolution semantics so callers may now receive multiple `ConstrainedValue`s for a key, which could subtly affect tests/components relying on last-writer-wins behavior. > > **Overview** > Adds *mergeable* dynamic config overrides to `MemoryClient` via `PartialOverrideSetting`/`PartialOverrideValue`, allowing multiple overrides for the same key to accumulate (intended for parallel tests with namespace-scoped constraints). > > Updates `GetValue` to scan overrides from newest to oldest, concatenating values from mergeable entries until a non-mergeable override is hit (which still stops the scan and shadows older overrides). Adds unit tests covering coexistence, mixed full vs partial override precedence, and non-LIFO removals. > > <sup>Written by [Cursor Bugbot]() for commit 7f1993835a7cc9e0e72eff0d1c6729aae0bb42c0. This will update automatically on new commits. Configure [here]().</sup>
119 lines
3.0 KiB
Go
119 lines
3.0 KiB
Go
package dynamicconfig
|
|
|
|
import (
|
|
"sync"
|
|
"sync/atomic"
|
|
)
|
|
|
|
type (
|
|
MemoryClient struct {
|
|
lock sync.RWMutex
|
|
overrides []kvpair
|
|
|
|
NotifyingClientImpl
|
|
}
|
|
|
|
kvpair struct {
|
|
valid bool
|
|
mergeable bool
|
|
key Key
|
|
value any
|
|
}
|
|
)
|
|
|
|
// NewMemoryClient - returns a memory based dynamic config client
|
|
func NewMemoryClient() *MemoryClient {
|
|
return &MemoryClient{NotifyingClientImpl: NewNotifyingClientImpl()}
|
|
}
|
|
|
|
func (d *MemoryClient) GetValue(key Key) []ConstrainedValue {
|
|
d.lock.RLock()
|
|
defer d.lock.RUnlock()
|
|
return d.getValueLocked(key)
|
|
}
|
|
|
|
func (d *MemoryClient) getValueLocked(key Key) []ConstrainedValue {
|
|
var result []ConstrainedValue
|
|
for i := len(d.overrides) - 1; i >= 0; i-- {
|
|
if d.overrides[i].valid && d.overrides[i].key == key {
|
|
v := d.overrides[i].value
|
|
if cvs, ok := v.([]ConstrainedValue); ok {
|
|
result = append(result, cvs...)
|
|
} else {
|
|
result = append(result, ConstrainedValue{Value: v})
|
|
}
|
|
if !d.overrides[i].mergeable {
|
|
// Non-mergeable: take this value and stop.
|
|
return result
|
|
}
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
func (d *MemoryClient) OverrideSetting(setting GenericSetting, value any) (cleanup func()) {
|
|
return d.OverrideValue(setting.Key(), value)
|
|
}
|
|
|
|
func (d *MemoryClient) OverrideValue(key Key, value any) (cleanup func()) {
|
|
return d.overrideValue(key, value, false)
|
|
}
|
|
|
|
// PartialOverrideSetting is like OverrideSetting but marks the override as mergeable.
|
|
// Mergeable overrides accumulate with other mergeable overrides for the same key,
|
|
// rather than shadowing them. This is used for namespace-constrained overrides on
|
|
// shared test clusters, where parallel tests need their overrides to coexist.
|
|
func (d *MemoryClient) PartialOverrideSetting(setting GenericSetting, value any) (cleanup func()) {
|
|
return d.PartialOverrideValue(setting.Key(), value)
|
|
}
|
|
|
|
// PartialOverrideValue is like OverrideValue but marks the override as mergeable.
|
|
func (d *MemoryClient) PartialOverrideValue(key Key, value any) (cleanup func()) {
|
|
return d.overrideValue(key, value, true)
|
|
}
|
|
|
|
func (d *MemoryClient) overrideValue(key Key, value any, mergeable bool) (cleanup func()) {
|
|
d.lock.Lock()
|
|
|
|
var idx atomic.Int64
|
|
idx.Store(int64(len(d.overrides)))
|
|
|
|
d.overrides = append(d.overrides, kvpair{valid: true, mergeable: mergeable, key: key, value: value})
|
|
|
|
newValue := d.getValueLocked(key)
|
|
changed := map[Key][]ConstrainedValue{key: newValue}
|
|
|
|
d.lock.Unlock()
|
|
|
|
// do not hold lock while notifying subscriptions
|
|
d.PublishUpdates(changed)
|
|
|
|
return func() {
|
|
// only do this once
|
|
if removeIdx := int(idx.Swap(-1)); removeIdx >= 0 {
|
|
d.remove(removeIdx)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (d *MemoryClient) remove(idx int) {
|
|
d.lock.Lock()
|
|
|
|
key := d.overrides[idx].key
|
|
// mark this pair deleted
|
|
d.overrides[idx] = kvpair{}
|
|
|
|
// pop all deleted pairs
|
|
for l := len(d.overrides); l > 0 && !d.overrides[l-1].valid; l = len(d.overrides) {
|
|
d.overrides = d.overrides[:l-1]
|
|
}
|
|
|
|
newValue := d.getValueLocked(key)
|
|
changed := map[Key][]ConstrainedValue{key: newValue}
|
|
|
|
d.lock.Unlock()
|
|
|
|
// do not hold lock while notifying subscriptions
|
|
d.PublishUpdates(changed)
|
|
}
|