m/pkg/event: move MemoryValue to subpackage
This keeps metropolis/pkg/event as a pure interface package, and
moves the memory-backed implementation to a subpackage.
Test Plan: Refactor, coevered by tests.
X-Origin-Diff: phab/D764
GitOrigin-RevId: 1337bf55a7752293791b3efe8648bbf5f6e6e9e1
diff --git a/metropolis/pkg/event/memory/BUILD.bazel b/metropolis/pkg/event/memory/BUILD.bazel
new file mode 100644
index 0000000..4ba79d2
--- /dev/null
+++ b/metropolis/pkg/event/memory/BUILD.bazel
@@ -0,0 +1,19 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+ name = "go_default_library",
+ srcs = ["memory.go"],
+ importpath = "source.monogon.dev/metropolis/pkg/event/memory",
+ visibility = ["//visibility:public"],
+ deps = ["//metropolis/pkg/event:go_default_library"],
+)
+
+go_test(
+ name = "go_default_test",
+ srcs = [
+ "example_test.go",
+ "memory_test.go",
+ ],
+ embed = [":go_default_library"],
+ deps = ["//metropolis/pkg/event:go_default_library"],
+)
diff --git a/metropolis/pkg/event/memory/example_test.go b/metropolis/pkg/event/memory/example_test.go
new file mode 100644
index 0000000..a119666
--- /dev/null
+++ b/metropolis/pkg/event/memory/example_test.go
@@ -0,0 +1,140 @@
+// Copyright 2020 The Monogon Project Authors.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package memory
+
+import (
+ "context"
+ "fmt"
+ "net"
+ "time"
+
+ "source.monogon.dev/metropolis/pkg/event"
+)
+
+// NetworkStatus is example data that will be stored in a Value.
+type NetworkStatus struct {
+ ExternalAddress net.IP
+ DefaultGateway net.IP
+}
+
+// NetworkStatusWatcher is a typesafe wrapper around a Watcher.
+type NetworkStatusWatcher struct {
+ watcher event.Watcher
+}
+
+// Get wraps Watcher.Get and performs type assertion.
+func (s *NetworkStatusWatcher) Get(ctx context.Context) (*NetworkStatus, error) {
+ val, err := s.watcher.Get(ctx)
+ if err != nil {
+ return nil, err
+ }
+ ns := val.(NetworkStatus)
+ return &ns, nil
+}
+
+// NetworkService is a fake/example network service that is responsible for
+// communicating the newest information about a machine's network configuration
+// to consumers/watchers.
+type NetworkService struct {
+ Provider Value
+}
+
+// Watch is a thin wrapper around Value.Watch that returns the typesafe
+// NetworkStatusWatcher wrapper.
+func (s *NetworkService) Watch() NetworkStatusWatcher {
+ watcher := s.Provider.Watch()
+ return NetworkStatusWatcher{
+ watcher: watcher,
+ }
+}
+
+// Run pretends to execute the network service's main logic loop, in which it
+// pretends to have received an IP address over DHCP, and communicates that to
+// consumers/watchers.
+func (s *NetworkService) Run(ctx context.Context) {
+ s.Provider.Set(NetworkStatus{
+ ExternalAddress: nil,
+ DefaultGateway: nil,
+ })
+
+ select {
+ case <-time.After(100 * time.Millisecond):
+ case <-ctx.Done():
+ return
+ }
+
+ fmt.Printf("NS: Got DHCP Lease\n")
+ s.Provider.Set(NetworkStatus{
+ ExternalAddress: net.ParseIP("203.0.113.24"),
+ DefaultGateway: net.ParseIP("203.0.113.1"),
+ })
+
+ select {
+ case <-time.After(100 * time.Millisecond):
+ case <-ctx.Done():
+ return
+ }
+
+ fmt.Printf("NS: DHCP Address changed\n")
+ s.Provider.Set(NetworkStatus{
+ ExternalAddress: net.ParseIP("203.0.113.103"),
+ DefaultGateway: net.ParseIP("203.0.113.1"),
+ })
+
+ time.Sleep(100 * time.Millisecond)
+}
+
+// ExampleValue_full demonstrates a typical usecase for Event Values, in which
+// a mock network service lets watchers know that the machine on which the code
+// is running has received a new network configuration.
+// It also shows the typical boilerplate required in order to wrap a Value (eg.
+// MemoryValue) within a typesafe wrapper.
+func ExampleValue_full() {
+ ctx, ctxC := context.WithCancel(context.Background())
+ defer ctxC()
+
+ // Create a fake NetworkService.
+ ns := NetworkService{}
+
+ // Run an /etc/hosts updater. It will watch for updates from the NetworkService
+ // about the current IP address of the node.
+ go func() {
+ w := ns.Watch()
+ for {
+ status, err := w.Get(ctx)
+ if err != nil {
+ break
+ }
+ if status.ExternalAddress == nil {
+ continue
+ }
+ // Pretend to write /etc/hosts with the newest ExternalAddress.
+ // In production code, you would also check for whether ExternalAddress has
+ // changed from the last written value, if writing to /etc/hosts is expensive.
+ fmt.Printf("/etc/hosts: foo.example.com is now %s\n", status.ExternalAddress.String())
+ }
+ }()
+
+ // Run fake network service.
+ ns.Run(ctx)
+
+ // Output:
+ // NS: Got DHCP Lease
+ // /etc/hosts: foo.example.com is now 203.0.113.24
+ // NS: DHCP Address changed
+ // /etc/hosts: foo.example.com is now 203.0.113.103
+}
diff --git a/metropolis/pkg/event/memory/memory.go b/metropolis/pkg/event/memory/memory.go
new file mode 100644
index 0000000..1e94a93
--- /dev/null
+++ b/metropolis/pkg/event/memory/memory.go
@@ -0,0 +1,282 @@
+// Copyright 2020 The Monogon Project Authors.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package memory
+
+import (
+ "context"
+ "fmt"
+ "sync"
+
+ "source.monogon.dev/metropolis/pkg/event"
+)
+
+var (
+ // Type assert that *Value implements Value. We do this artificially, as
+ // there currently is no code path that needs this to be strictly true. However,
+ // users of this library might want to rely on the Value type instead of
+ // particular Value implementations.
+ _ event.Value = &Value{}
+)
+
+// Value is a 'memory value', which implements a event.Value stored in memory.
+// It is safe to construct an empty object of this type. However, this must not
+// be copied.
+type Value struct {
+ // mu guards the inner, innerSet and watchers fields.
+ mu sync.RWMutex
+ // inner is the latest data Set on the Value. It is used to provide the
+ // newest version of the Set data to new watchers.
+ inner interface{}
+ // innerSet is true when inner has been Set at least once. It is used to
+ // differentiate between a nil and unset value.
+ innerSet bool
+ // watchers is the list of watchers that should be updated when new data is
+ // Set. It will grow on every .Watch() and shrink any time a watcher is
+ // determined to have been closed.
+ watchers []*watcher
+
+ // Sync, if set to true, blocks all .Set() calls on the Value until all
+ // Watchers derived from it actively .Get() the new value. This can be used
+ // to ensure Watchers always receive a full log of all Set() calls.
+ //
+ // This must not be changed after the first .Set/.Watch call.
+ //
+ // This is an experimental API and subject to change. It might be migrated
+ // to per-Watcher settings defined within the main event.Value/Watcher
+ // interfaces.
+ Sync bool
+}
+
+// Set updates the Value to the given data. It is safe to call this from
+// multiple goroutines, including concurrently.
+//
+// For more information about guarantees, see event.Value.Set.
+func (m *Value) Set(val interface{}) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ // Update the data that is provided on first Get() to watchers.
+ m.inner = val
+ m.innerSet = true
+
+ // Go through all watchers, updating them on the new value and filtering out
+ // all closed watchers.
+ newWatchers := make([]*watcher, 0, len(m.watchers))
+ for _, w := range m.watchers {
+ w := w
+ if w.closed() {
+ continue
+ }
+ w.update(m.Sync, val)
+ newWatchers = append(newWatchers, w)
+ }
+ m.watchers = newWatchers
+}
+
+// watcher implements the event.Watcher interface for watchers returned by
+// Value.
+type watcher struct {
+ // activeReqC is a channel used to request an active submission channel
+ // from a pending Get function, if any.
+ activeReqC chan chan interface{}
+ // deadletterSubmitC is a channel used to communicate a value that
+ // attempted to be submitted via activeReqC. This will be received by the
+ // deadletter worker of this watcher and passed on to the next .Get call
+ // that occurs.
+ deadletterSubmitC chan interface{}
+
+ // getSem is a channel-based semaphore (which is of size 1, and thus in
+ // fact a mutex) that is used to ensure that only a single .Get() call is
+ // active. It is implemented as a channel to permit concurrent .Get() calls
+ // to error out instead of blocking.
+ getSem chan struct{}
+ // close is a channel that is closed when this watcher is itself Closed.
+ close chan struct{}
+}
+
+// Watch retrieves a Watcher that keeps track on the version of the data
+// contained within the Value that was last seen by a consumer.
+//
+// For more information about guarantees, see event.Value.Watch.
+func (m *Value) Watch() event.Watcher {
+ waiter := &watcher{
+ activeReqC: make(chan chan interface{}),
+ deadletterSubmitC: make(chan interface{}),
+ close: make(chan struct{}),
+ getSem: make(chan struct{}, 1),
+ }
+ // Start the deadletter worker as a goroutine. It will be stopped when the
+ // watcher is Closed() (as signaled by the close channel).
+ go waiter.deadletterWorker()
+
+ // Append this watcher to the Value.
+ m.mu.Lock()
+ m.watchers = append(m.watchers, waiter)
+ // If the Value already has some value set, communicate that to the
+ // first Get call by going through the deadletter worker.
+ if m.innerSet {
+ waiter.deadletterSubmitC <- m.inner
+ }
+ m.mu.Unlock()
+
+ return waiter
+}
+
+// deadletterWorker runs the 'deadletter worker', as goroutine that contains
+// any data that has been Set on the Value that is being watched that was
+// unable to be delivered directly to a pending .Get call.
+//
+// It watches the deadletterSubmitC channel for updated data, and overrides
+// previously received data. Then, when a .Get() begins to pend (and respond to
+// activeReqC receives), the deadletter worker will deliver that value.
+func (m *watcher) deadletterWorker() {
+ // Current value, and flag to mark it as set (vs. nil).
+ var cur interface{}
+ var set bool
+
+ for {
+ if !set {
+ // If no value is yet available, only attempt to receive one from the
+ // submit channel, as there's nothing to submit to pending .Get() calls
+ // yet.
+ val, ok := <-m.deadletterSubmitC
+ if !ok {
+ // If the channel has been closed (by Close()), exit.
+ return
+ }
+ cur = val
+ set = true
+ } else {
+ // If a value is available, update the inner state. Otherwise, if a
+ // .Get() is pending, submit our current state and unset it.
+ select {
+ case val, ok := <-m.deadletterSubmitC:
+ if !ok {
+ // If the channel has been closed (by Close()), exit.
+ return
+ }
+ cur = val
+ case c := <-m.activeReqC:
+ // Potential race: a .Get() might've been active, but might've
+ // quit by the time we're here (and will not receive on the
+ // responded channel). Handle this gracefully by just returning
+ // to the main loop if that's the case.
+ select {
+ case c <- cur:
+ set = false
+ default:
+ }
+ }
+ }
+ }
+}
+
+// closed returns whether this watcher has been closed.
+func (m *watcher) closed() bool {
+ select {
+ case _, ok := <-m.close:
+ if !ok {
+ return true
+ }
+ default:
+ }
+ return false
+}
+
+// update is the high level update-this-watcher function called by Value.
+func (m *watcher) update(sync bool, val interface{}) {
+ // If synchronous delivery was requested, block until a watcher .Gets it.
+ if sync {
+ c := <-m.activeReqC
+ c <- val
+ return
+ }
+
+ // Otherwise, deliver asynchronously. This means either delivering directly
+ // to a pending .Get if one exists, or submitting to the deadletter worker
+ // otherwise.
+ select {
+ case c := <-m.activeReqC:
+ // Potential race: a .Get() might've been active, but might've quit by
+ // the time we're here (and will not receive on the responded channel).
+ // Handle this gracefully by falling back to the deadletter worker.
+ select {
+ case c <- val:
+ default:
+ m.deadletterSubmitC <- val
+ }
+ default:
+ m.deadletterSubmitC <- val
+ }
+}
+
+func (m *watcher) Close() error {
+ close(m.deadletterSubmitC)
+ close(m.close)
+ return nil
+}
+
+// Get blocks until a Value's data is available. See event.Watcher.Get for
+// guarantees and more information.
+func (m *watcher) Get(ctx context.Context) (interface{}, error) {
+ // Make sure we're the only active .Get call.
+ select {
+ case m.getSem <- struct{}{}:
+ default:
+ return nil, fmt.Errorf("cannot Get() concurrently on a single waiter")
+ }
+ defer func() {
+ <-m.getSem
+ }()
+
+ c := make(chan interface{})
+
+ // Start responding on activeReqC. This signals to .update and to the
+ // deadletter worker that we're ready to accept data updates.
+
+ // There is a potential for a race condition here that hasn't been observed
+ // in tests but might happen:
+ // 1) Value.Watch returns a Watcher 'w'.
+ // 2) w.Set(0) is called, no .Get() is pending, so 0 is submitted to the
+ // deadletter worker.
+ // 3) w.Get() is called, and activeReqC begins to be served.
+ // 4) Simultaneously:
+ // a) w.Set(1) is called, attempting to submit via activeReqC
+ // b) the deadletter worker attempts to submit via activeReqC
+ //
+ // This could theoretically cause .Get() to first return 1, and then 0, if
+ // the Set activeReqC read and subsequent channel write is served before
+ // the deadletter workers' read/write is.
+ // As noted, however, this has not been observed in practice, even though
+ // TestConcurrency explicitly attempts to trigger this condition. More
+ // research needs to be done to attempt to trigger this (or to lawyer the
+ // Go channel spec to see if this has some guarantees that resolve this
+ // either way), or a preemptive fix can be attempted by adding monotonic
+ // counters associated with each .Set() value, ensuring an older value does
+ // not replace a newer value.
+ //
+ // TODO(q3k): investigate this.
+ for {
+ select {
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ case m.activeReqC <- c:
+ case val := <-c:
+ return val, nil
+ }
+ }
+}
diff --git a/metropolis/pkg/event/memory/memory_test.go b/metropolis/pkg/event/memory/memory_test.go
new file mode 100644
index 0000000..f4feb33
--- /dev/null
+++ b/metropolis/pkg/event/memory/memory_test.go
@@ -0,0 +1,332 @@
+// Copyright 2020 The Monogon Project Authors.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package memory
+
+import (
+ "context"
+ "fmt"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+// TestAsync exercises the high-level behaviour of a Value, in which a
+// watcher is able to catch up to the newest Set value.
+func TestAsync(t *testing.T) {
+ p := Value{}
+ p.Set(0)
+
+ ctx := context.Background()
+
+ // The 0 from Set() should be available via .Get().
+ watcher := p.Watch()
+ val, err := watcher.Get(ctx)
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ if want, got := 0, val.(int); want != got {
+ t.Fatalf("Value: got %d, wanted %d", got, want)
+ }
+
+ // Send a large amount of updates that the watcher does not actively .Get().
+ for i := 1; i <= 100; i++ {
+ p.Set(i)
+ }
+
+ // The watcher should still end up with the newest .Set() value on the next
+ // .Get() call.
+ val, err = watcher.Get(ctx)
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ if want, got := 100, val.(int); want != got {
+ t.Fatalf("Value: got %d, wanted %d", got, want)
+ }
+}
+
+// TestSyncBlocks exercises the Value's 'Sync' field, which makes all
+// Set() calls block until all respective watchers .Get() the updated data.
+// This particular test ensures that .Set() calls to a Watcher result in a
+// prefect log of updates being transmitted to a watcher.
+func TestSync(t *testing.T) {
+ p := Value{
+ Sync: true,
+ }
+ values := make(chan int, 100)
+ wg := sync.WaitGroup{}
+ wg.Add(1)
+ go func() {
+ ctx := context.Background()
+ watcher := p.Watch()
+ wg.Done()
+ for {
+ value, err := watcher.Get(ctx)
+ if err != nil {
+ panic(err)
+ }
+ values <- value.(int)
+ }
+ }()
+
+ p.Set(0)
+ wg.Wait()
+
+ want := []int{1, 2, 3, 4}
+ for _, w := range want {
+ p.Set(w)
+ }
+
+ timeout := time.After(time.Second)
+ for i, w := range append([]int{0}, want...) {
+ select {
+ case <-timeout:
+ t.Fatalf("timed out on value %d (%d)", i, w)
+ case val := <-values:
+ if w != val {
+ t.Errorf("value %d was %d, wanted %d", i, val, w)
+ }
+ }
+ }
+}
+
+// TestSyncBlocks exercises the Value's 'Sync' field, which makes all
+// Set() calls block until all respective watchers .Get() the updated data.
+// This particular test ensures that .Set() calls actually block when a watcher
+// is unattended.
+func TestSyncBlocks(t *testing.T) {
+ p := Value{
+ Sync: true,
+ }
+ ctx := context.Background()
+
+ // Shouldn't block, as there's no declared watchers.
+ p.Set(0)
+
+ watcher := p.Watch()
+
+ // Should retrieve the zero, more requests will pend.
+ value, err := watcher.Get(ctx)
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ if want, got := 0, value.(int); want != got {
+ t.Fatalf("Got initial value %d, wanted %d", got, want)
+ }
+
+ // .Set() Should block, as watcher is unattended.
+ //
+ // Whether something blocks in Go is untestable in a robust way (see: halting
+ // problem). We work around this this by introducing a 'stage' int64, which is
+ // put on the 'c' channel after the needs-to-block function returns. We then
+ // perform an action that should unblock this function right after updating
+ // 'stage' to a different value.
+ // Then, we observe what was put on the channel: If it's the initial value, it
+ // means the function didn't block when expected. Otherwise, it means the
+ // function unblocked when expected.
+ stage := int64(0)
+ c := make(chan int64, 1)
+ go func() {
+ p.Set(1)
+ c <- atomic.LoadInt64(&stage)
+ }()
+
+ // Getting should unblock the provider. Mark via 'stage' variable that
+ // unblocking now is expected.
+ atomic.StoreInt64(&stage, int64(1))
+ // Potential race: .Set() unblocks here due to some bug, before .Get() is
+ // called, and we record a false positive.
+ value, err = watcher.Get(ctx)
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+
+ res := <-c
+ if res != int64(1) {
+ t.Fatalf("Set() returned before Get()")
+ }
+
+ if want, got := 1, value.(int); want != got {
+ t.Fatalf("Wanted value %d, got %d", want, got)
+ }
+
+ // Closing the watcher and setting should not block anymore.
+ if err := watcher.Close(); err != nil {
+ t.Fatalf("Close: %v", err)
+ }
+ // Last step, if this blocks we will get a deadlock error and the test will panic.
+ p.Set(2)
+}
+
+// TestMultipleGets verifies that calling .Get() on a single watcher from two
+// goroutines is prevented by returning an error in exactly one of them.
+func TestMultipleGets(t *testing.T) {
+ p := Value{}
+ ctx := context.Background()
+
+ w := p.Watch()
+
+ tryError := func(errs chan error) {
+ _, err := w.Get(ctx)
+ errs <- err
+ }
+ errs := make(chan error, 2)
+ go tryError(errs)
+ go tryError(errs)
+
+ for err := range errs {
+ if err == nil {
+ t.Fatalf("A Get call succeeded, while it should have blocked or returned an error")
+ } else {
+ // Found the error, test succeeded.
+ break
+ }
+ }
+}
+
+// TestConcurrency attempts to stress the Value/Watcher
+// implementation to design limits (a hundred simultaneous watchers), ensuring
+// that the watchers all settle to the final set value.
+func TestConcurrency(t *testing.T) {
+ ctx := context.Background()
+
+ p := Value{}
+ p.Set(0)
+
+ // Number of watchers to create.
+ watcherN := 100
+ // Expected final value to be Set().
+ final := 100
+ // Result channel per watcher.
+ resC := make([]chan error, watcherN)
+
+ // Spawn watcherN watchers.
+ for i := 0; i < watcherN; i++ {
+ resC[i] = make(chan error, 1)
+ go func(id int) {
+ // done is a helper function that will put an error on the
+ // respective watcher's resC.
+ done := func(err error) {
+ resC[id] <- err
+ close(resC[id])
+ }
+
+ watcher := p.Watch()
+ // prev is used to ensure the values received are monotonic.
+ prev := -1
+ for {
+ val, err := watcher.Get(ctx)
+ if err != nil {
+ done(err)
+ return
+ }
+
+ // Ensure monotonicity of received data.
+ if val.(int) <= prev {
+ done(fmt.Errorf("received out of order data: %d after %d", val, prev))
+ }
+ prev = val.(int)
+
+ // Quit when the final value is received.
+ if val == final {
+ done(nil)
+ return
+ }
+
+ // Sleep a bit, depending on the watcher. This makes each
+ // watcher behave slightly differently, and attempts to
+ // exercise races dependent on sleep time between subsequent
+ // Get calls.
+ time.Sleep(time.Millisecond * time.Duration(id))
+ }
+ }(i)
+ }
+
+ // Set 1..final on the value.
+ for i := 1; i <= final; i++ {
+ p.Set(i)
+ }
+
+ // Ensure all watchers exit with no error.
+ for i, c := range resC {
+ err := <-c
+ if err != nil {
+ t.Errorf("Watcher %d returned %v", i, err)
+ }
+ }
+}
+
+// TestCanceling exercises whether a context canceling in a .Get() gracefully
+// aborts that particular Get call, but also allows subsequent use of the same
+// watcher.
+func TestCanceling(t *testing.T) {
+ p := Value{
+ Sync: true,
+ }
+
+ ctx, ctxC := context.WithCancel(context.Background())
+
+ watcher := p.Watch()
+
+ // errs will contain the error returned by Get.
+ errs := make(chan error, 1)
+ go func() {
+ // This Get will block, as no initial data has been Set on the value.
+ _, err := watcher.Get(ctx)
+ errs <- err
+ }()
+
+ // Cancel the context, and expect that context error to propagate to the .Get().
+ ctxC()
+ if want, got := ctx.Err(), <-errs; want != got {
+ t.Fatalf("Get should've returned %v, got %v", want, got)
+ }
+
+ // Do another .Get() on the same watcher with a new context. Even though the
+ // call was aborted via a context cancel, the watcher should continue working.
+ ctx = context.Background()
+ go func() {
+ _, err := watcher.Get(ctx)
+ errs <- err
+ }()
+
+ // Unblock the .Get now.
+ p.Set(1)
+ if want, got := error(nil), <-errs; want != got {
+ t.Fatalf("Get should've returned %v, got %v", want, got)
+ }
+}
+
+// TestSetAfterWatch ensures that if a value is updated between a Watch and the
+// initial Get, only the newest Set value is returns.
+func TestSetAfterWatch(t *testing.T) {
+ ctx := context.Background()
+
+ p := Value{}
+ p.Set(0)
+
+ watcher := p.Watch()
+ p.Set(1)
+
+ data, err := watcher.Get(ctx)
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ if want, got := 1, data.(int); want != got {
+ t.Errorf("Get should've returned %v, got %v", want, got)
+ }
+}