diff --git a/metropolis/pkg/event/BUILD.bazel b/metropolis/pkg/event/BUILD.bazel
new file mode 100644
index 0000000..4e3b33c
--- /dev/null
+++ b/metropolis/pkg/event/BUILD.bazel
@@ -0,0 +1,20 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "event.go",
+        "memory_value.go",
+    ],
+    importpath = "source.monogon.dev/metropolis/pkg/event",
+    visibility = ["//visibility:public"],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = [
+        "example_test.go",
+        "memory_value_test.go",
+    ],
+    embed = [":go_default_library"],
+)
diff --git a/metropolis/pkg/event/event.go b/metropolis/pkg/event/event.go
new file mode 100644
index 0000000..d3ea5df
--- /dev/null
+++ b/metropolis/pkg/event/event.go
@@ -0,0 +1,166 @@
+// 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 event defines and implements Event Values, a mechanism in which
+// multiple consumers can watch a value for updates in a reliable way.
+//
+// Values currently are kept in memory (see: MemoryValue), but a future
+// implementation might exist for other storage backends, eg. etcd.
+//
+// Background and intended use
+//
+// The Event Value library is intended to be used within Metropolis'
+// supervisor-based runnables to communicate state changes to other runnables,
+// while permitting both sides to restart if needed. It grew out of multiple
+// codebases reimplementing an ad-hoc observer pattern, and from the
+// realization that implementing all possible edge cases of such patterns is
+// non-trivial and subject to programming errors. As such, it was turned into a
+// self-standing library.
+//
+// Why not just channels?
+//
+// Plain channels have multiple deficiencies for this usecase:
+//  - Strict FIFO behaviour: all values sent to a channel must be received, and
+//    historic and newest data must be treated in the same way. This means that
+//    a consumer of state changes must process all updates to the value as if
+//    they are the newest, and unable to skip rapid updates when a system is
+//    slowly settling due to a cascading state change.
+//  - Implementation overhead: implementing an observer
+//    registration/unregistration pattern is prone to programming bugs,
+//    especially for features like always first sending the current state to a
+//    new observer.
+//  - Strict buffer size: due to their FIFO nature and the possibility of
+//    consumers not receiving actively, channels would have to buffer all
+//    existing updates, requiring some arbitrary best-guess channel buffer
+//    sizing that would still not prevent blocking writes or data loss in a
+//    worst case scenario.
+//
+// Or, in other words: Go channels are a synchronization primitive, not a
+// ready-made solution to this problem. The Event Value implementation in fact
+// extensively uses Go channels within its implementation as a building block.
+//
+// Why not just condition variables (sync.Cond)?
+//
+// Go's condition variable implementation doesn't fully address our needs
+// either:
+// - No context/canceling support: once a condition is being Wait()ed on,
+//   this cannot be interrupted. This is especially painful and unwieldy when
+//   dealing with context-heavy code, such as Metropolis.
+// - Spartan API: expecting users to plainly use sync.Cond is risky, as the API
+//   is fairly low-level.
+// - No solution for late consumers: late consumers (ones that missed the value
+//   being set by a producer) would still have to implement logic in order to
+//   find out such a value, as sync.Cond only supports what amounts to
+//   edge-level triggers as part of its Broadcast/Signal system.
+//
+// It would be possible to implement MemoryValue using a sync.Cond internally,
+// but such an implementation would likely be more complex than the current
+// implementation based on channels and mutexes, as it would have to work
+// around issues like lack of canceling, etc.
+//
+// Type safety
+//
+// The Value/Watcher interfaces are, unfortunately, implemented using
+// interface{}. There was an attempt to use Go's existing generic types facility
+// (interfaces) to solve this problem. However, with Type Parameters likely soon
+// appearing in mainline Go, this was not a priority, as that will fully solve
+// this problem without requiring mental gymnastics. For now, users of this
+// library will have to write some boilerplate code to allow consumers/watchers
+// to access the data in a a typesafe manner without assertions. See
+// ExampleValue_full for one possible approach to this.
+package event
+
+import (
+	"context"
+)
+
+// A Value is an 'Event Value', some piece of data that can be updated ('Set')
+// by Producers and retrieved by Consumers.
+type Value interface {
+	// Set updates the Value to the given data. It is safe to call this from
+	// multiple goroutines, including concurrently.
+	//
+	// Any time Set is called, any consumers performing a Watch on this Value
+	// will be notified with the new data - even if the Set data is the same as
+	// the one that was already stored.
+	//
+	// A Value will initially have no data set. This 'no data' state is seen by
+	// consumers by the first .Get() call on the Watcher blocking until data is Set.
+	//
+	// All updates will be serialized in an arbitrary order - if multiple
+	// producers wish to perform concurrent actions to update the Value partially,
+	// this should be negotiated and serialized externally by the producers.
+	Set(val interface{})
+
+	// Watch retrieves a Watcher that keeps track on the version of the data
+	// contained within the Value that was last seen by a consumer. Once a
+	// Watcher is retrieved, it can be used to then get the actual data stored
+	// within the Value, and to reliably retrieve updates to it without having
+	// to poll for changes.
+	Watch() Watcher
+}
+
+// A Watcher keeps track of the last version of data seen by a consumer for a
+// given Value. Each consumer should use an own Watcher instance, and it is not
+// safe to use this type concurrently. However, it is safe to move/copy it
+// across different goroutines, as long as no two goroutines access it
+// simultaneously.
+type Watcher interface {
+	// Get blocks until a Value's data is available:
+	//  - On first use of a Watcher, Get will return the data contained in the
+	//    value at the time of calling .Watch(), or block if no data has been
+	//    .Set() on it yet. If a value has been Set() since the the initial
+	//    creation of the Watch() but before Get() is called for the first
+	//    time, the first Get() call will immediately return the new value.
+	//  - On subsequent uses of a Watcher, Get will block until the given Value
+	//    has been Set with new data. This does not necessarily mean that the
+	//    new data is different - consumers should always perform their own
+	//    checks on whether the update is relevant to them (ie., the data has
+	//    changed in a significant way), unless specified otherwise by a Value
+	//    publisher.
+	//
+	// Get() will always return the current newest data that has been Set() on
+	// the Value, and not a full log of historical events. This is geared
+	// towards event values where consumers only care about changes to data
+	// since last retrieval, not every value that has been Set along the way.
+	// Thus, consumers need not make sure that they actively .Get() on a
+	// watcher all the times.
+	//
+	// If the context is canceled before data is available to be returned, the
+	// context's error will be returned. However, the Watcher will still need to be
+	// Closed, as it is still fully functional after the context has been canceled.
+	//
+	// Concurrent requests to Get result in an error. The reasoning to return
+	// an error instead of attempting to serialize the requests is that any
+	// concurrent access from multiple goroutines would cause a desync in the
+	// next usage of the Watcher. For example:
+	//   1) w.Get() (in G0) and w.Get(G1) start. They both block waiting for an
+	//      initial value.
+	//   2) v.Set(0)
+	//   3) w.Get() in G0 returns 0,
+	//   4) v.Set(1)
+	//   4) w.Get() in G1 returns 1,
+	// This would cause G0 and G1 to become desynchronized between eachother
+	// (both have different value data) and subsequent updates will also
+	// continue skipping some updates.
+	// If multiple goroutines need to access the Value, they should each use
+	// their own Watcher.
+	Get(context.Context) (interface{}, error)
+
+	// Close must be called if the Watcher is not going to be used anymore -
+	// otherwise, a goroutine will leak.
+	Close() error
+}
diff --git a/metropolis/pkg/event/example_test.go b/metropolis/pkg/event/example_test.go
new file mode 100644
index 0000000..3cdc661
--- /dev/null
+++ b/metropolis/pkg/event/example_test.go
@@ -0,0 +1,139 @@
+// 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 event
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+)
+
+// 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 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 MemoryValue
+}
+
+// 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_value.go b/metropolis/pkg/event/memory_value.go
new file mode 100644
index 0000000..adf412d
--- /dev/null
+++ b/metropolis/pkg/event/memory_value.go
@@ -0,0 +1,280 @@
+// 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 event
+
+import (
+	"context"
+	"fmt"
+	"sync"
+)
+
+var (
+	// Type assert that *MemoryValue 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.
+	_ Value = &MemoryValue{}
+)
+
+// MemoryValue implements a Value stored in memory. It is safe to construct an
+// empty object of this type. However, this must not be copied.
+type MemoryValue struct {
+	// mu guards the inner, innerSet and watchers fields.
+	mu sync.RWMutex
+	// inner is the latest data Set on the MemoryValue. 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 []*MemoryWatcher
+
+	// Sync, if set to true, blocks all .Set() calls on the MemoryValue 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 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 Value.Set.
+func (m *MemoryValue) 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([]*MemoryWatcher, 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
+}
+
+// MemoryWatcher implements the Watcher interface for watchers returned by
+// MemoryValue.
+type MemoryWatcher 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 Value.Watch.
+func (m *MemoryValue) Watch() Watcher {
+	waiter := &MemoryWatcher{
+		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 MemoryValue.
+	m.mu.Lock()
+	m.watchers = append(m.watchers, waiter)
+	// If the MemoryValue 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 *MemoryWatcher) 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 *MemoryWatcher) 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 MemoryValue.
+func (m *MemoryWatcher) 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 *MemoryWatcher) Close() error {
+	close(m.deadletterSubmitC)
+	close(m.close)
+	return nil
+}
+
+// Get blocks until a Value's data is available. See Watcher.Get for guarantees
+// and more information.
+func (m *MemoryWatcher) 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_value_test.go b/metropolis/pkg/event/memory_value_test.go
new file mode 100644
index 0000000..4b0487c
--- /dev/null
+++ b/metropolis/pkg/event/memory_value_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 event
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// TestAsync exercises the high-level behaviour of a MemoryValue, in which a
+// watcher is able to catch up to the newest Set value.
+func TestAsync(t *testing.T) {
+	p := MemoryValue{}
+	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 MemoryValue'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 := MemoryValue{
+		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 MemoryValue'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 := MemoryValue{
+		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 := MemoryValue{}
+	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 MemoryValue/MemoryWatcher
+// 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 := MemoryValue{}
+	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 := MemoryValue{
+		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 := MemoryValue{}
+	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)
+	}
+}