Skip to content

Commit df46cdb

Browse files
mnoah1mnoah1
committed
fix(stovepipe): initialize browser test databases before startup
Apply schemas and fixtures before starting services, and cover startup order and cleanup in the runner test. Add queue and failure context to enablement and history errors while preserving sentinel matching.
1 parent 61a852f commit df46cdb

5 files changed

Lines changed: 74 additions & 36 deletions

File tree

‎stovepipe/core/queuelog/materializer.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,7 @@ func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, l
5555
return err
5656
}
5757
if err := log.Validate(); err != nil {
58-
return fmt.Errorf("%w: %v", ErrInconsistentHistory, err)
58+
return fmt.Errorf("queue %q: invalid log %q at state version %d: %w: %v", log.Queue, log.ID, log.QueueStateVersion, ErrInconsistentHistory, err)
5959
}
6060
logs := stores.GetQueueLogStore()
6161
if err := logs.Create(ctx, log); err != nil {
@@ -67,7 +67,7 @@ func (m *materializer) PersistLog(ctx context.Context, stores storage.Storage, l
6767
return fmt.Errorf("reconcile queue log: %w", err)
6868
}
6969
if retained != log {
70-
return ErrInconsistentHistory
70+
return fmt.Errorf("queue %q: log %q at state version %d conflicts with retained log %q for queue %q at state version %d: %w", log.Queue, log.ID, log.QueueStateVersion, retained.ID, retained.Queue, retained.QueueStateVersion, ErrInconsistentHistory)
7171
}
7272
log = retained
7373
}
@@ -94,22 +94,22 @@ func materializeQueueSummary(ctx context.Context, store storage.QueueSummaryStor
9494
return err
9595
}
9696
if current.Queue != log.Queue {
97-
return ErrInconsistentHistory
97+
return fmt.Errorf("queue %q log %q: summary belongs to queue %q at projection version %d: %w", log.Queue, log.ID, current.Queue, current.Version, ErrInconsistentHistory)
9898
}
9999
if err := current.Validate(); err != nil {
100-
return fmt.Errorf("%w: %v", ErrInconsistentHistory, err)
100+
return fmt.Errorf("queue %q: invalid summary at projection version %d, state version %d: %w: %v", log.Queue, current.Version, current.QueueStateVersion, ErrInconsistentHistory, err)
101101
}
102102
if log.QueueStateVersion < current.QueueStateVersion {
103103
return nil
104104
}
105105
if log.QueueStateVersion == current.QueueStateVersion {
106106
if current.Snapshot() != log.Snapshot() {
107-
return ErrInconsistentHistory
107+
return fmt.Errorf("queue %q: log %q conflicts with summary state version %d (projection version %d): %w", log.Queue, log.ID, current.QueueStateVersion, current.Version, ErrInconsistentHistory)
108108
}
109109
return nil
110110
}
111111
if current.Version == math.MaxInt64 {
112-
return ErrInconsistentHistory
112+
return fmt.Errorf("queue %q log %q: summary projection version %d exhausted: %w", log.Queue, log.ID, current.Version, ErrInconsistentHistory)
113113
}
114114
updated := current
115115
updated.State, updated.QueueStateVersion = log.State, log.QueueStateVersion

‎stovepipe/core/queuelog/queue_summary.go‎

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -32,35 +32,36 @@ func LoadSummary(ctx context.Context, stores storage.Storage, queue string) (ent
3232
return entity.QueueState{}, err
3333
}
3434
if summary.Queue != queue {
35-
return entity.QueueState{}, ErrInconsistentHistory
35+
return entity.QueueState{}, fmt.Errorf("queue %q: summary belongs to queue %q at projection version %d: %w", queue, summary.Queue, summary.Version, ErrInconsistentHistory)
3636
}
3737
if err := summary.Validate(); err != nil {
38-
return entity.QueueState{}, fmt.Errorf("%w: %v", ErrInconsistentHistory, err)
38+
return entity.QueueState{}, fmt.Errorf("queue %q: invalid summary at projection version %d, state version %d: %w: %v", queue, summary.Version, summary.QueueStateVersion, ErrInconsistentHistory, err)
3939
}
4040
retained, err := LoadVersion(ctx, stores.GetQueueLogStore(), queue, summary.QueueStateVersion)
4141
if err != nil {
4242
return entity.QueueState{}, err
4343
}
4444
if retained != summary.Snapshot() {
45-
return entity.QueueState{}, ErrInconsistentHistory
45+
return entity.QueueState{}, fmt.Errorf("queue %q: summary state version %d at projection version %d conflicts with retained history: %w", queue, summary.QueueStateVersion, summary.Version, ErrInconsistentHistory)
4646
}
4747
return summary.Snapshot(), nil
4848
}
4949

5050
// LoadVersion returns ErrInconsistentHistory for missing or invalid history.
5151
func LoadVersion(ctx context.Context, logs storage.QueueLogStore, queue string, version int64) (entity.QueueState, error) {
52-
log, err := logs.Get(ctx, entity.QueueLogID(version))
52+
logID := entity.QueueLogID(version)
53+
log, err := logs.Get(ctx, logID)
5354
if err != nil {
5455
if storage.IsNotFound(err) {
55-
return entity.QueueState{}, fmt.Errorf("missing state version %d: %w", version, ErrInconsistentHistory)
56+
return entity.QueueState{}, fmt.Errorf("queue %q: missing log %q for state version %d: %w", queue, logID, version, ErrInconsistentHistory)
5657
}
5758
return entity.QueueState{}, err
5859
}
5960
if log.Queue != queue || log.QueueStateVersion != version {
60-
return entity.QueueState{}, ErrInconsistentHistory
61+
return entity.QueueState{}, fmt.Errorf("queue %q: log %q for state version %d belongs to queue %q at state version %d: %w", queue, logID, version, log.Queue, log.QueueStateVersion, ErrInconsistentHistory)
6162
}
6263
if err := log.Validate(); err != nil {
63-
return entity.QueueState{}, fmt.Errorf("%w: %v", ErrInconsistentHistory, err)
64+
return entity.QueueState{}, fmt.Errorf("queue %q: invalid log %q at state version %d: %w: %v", queue, logID, version, ErrInconsistentHistory, err)
6465
}
6566
return log.Snapshot(), nil
6667
}

‎stovepipe/core/queuestate/queuestate.go‎

Lines changed: 15 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,7 @@ func (c *transitionController) Initialize(ctx context.Context, name string) (ent
7575
}
7676
if !missing {
7777
if queue.Name != name || queue.Version <= 0 {
78-
return entity.QueueState{}, queuelog.ErrInconsistentHistory
78+
return entity.QueueState{}, fmt.Errorf("initialize queue %q: invalid stored queue %q at row version %d: %w", name, queue.Name, queue.Version, queuelog.ErrInconsistentHistory)
7979
}
8080
if queue.EnablementVersion != 0 || queue.Enablement != entity.QueueEnablementUnknown || queue.EffectiveFromCommitURI != "" || queue.EnablementChangedAtMs != 0 {
8181
return c.repairQueueReadModels(ctx, stores, queue)
@@ -87,7 +87,7 @@ func (c *transitionController) Initialize(ctx context.Context, name string) (ent
8787
updated.Enablement, updated.EnablementVersion = entity.QueueEnablementDisabled, 1
8888
updated.EnablementChangedAtMs = c.now().UnixMilli()
8989
if err := updated.EnablementState().Validate(); err != nil {
90-
return entity.QueueState{}, ErrInvalidChange
90+
return entity.QueueState{}, fmt.Errorf("initialize queue %q: invalid enablement: %w: %v", name, ErrInvalidChange, err)
9191
}
9292
if missing {
9393
if err := store.Create(ctx, updated); errors.Is(err, storage.ErrAlreadyExists) {
@@ -97,7 +97,7 @@ func (c *transitionController) Initialize(ctx context.Context, name string) (ent
9797
}
9898
} else {
9999
if queue.Version == math.MaxInt32 {
100-
return entity.QueueState{}, ErrInvalidChange
100+
return entity.QueueState{}, fmt.Errorf("initialize queue %q: row version %d exhausted: %w", name, queue.Version, ErrInvalidChange)
101101
}
102102
newVersion := queue.Version + 1
103103
if err := store.Update(ctx, updated, queue.Version, newVersion); errors.Is(err, storage.ErrVersionMismatch) {
@@ -113,7 +113,7 @@ func (c *transitionController) Initialize(ctx context.Context, name string) (ent
113113

114114
func (c *transitionController) Prepare(ctx context.Context, queue string, enablement entity.QueueEnablement) (PreparedChange, error) {
115115
if enablement != entity.QueueEnablementEnabled && enablement != entity.QueueEnablementDisabled {
116-
return PreparedChange{}, ErrInvalidChange
116+
return PreparedChange{}, fmt.Errorf("prepare queue %q: unsupported enablement %q: %w", queue, enablement, ErrInvalidChange)
117117
}
118118
stores, err := c.resolveStores(ctx, queue)
119119
if err != nil {
@@ -129,7 +129,7 @@ func (c *transitionController) Prepare(ctx context.Context, queue string, enable
129129
return change, nil
130130
}
131131
if current.Version == math.MaxInt64 {
132-
return PreparedChange{}, ErrInvalidChange
132+
return PreparedChange{}, fmt.Errorf("prepare queue %q: enablement version %d exhausted: %w", queue, current.Version, ErrInvalidChange)
133133
}
134134
sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: queue})
135135
if err != nil {
@@ -140,28 +140,28 @@ func (c *transitionController) Prepare(ctx context.Context, queue string, enable
140140
return PreparedChange{}, err
141141
}
142142
if boundary == "" || len(boundary) > 255 {
143-
return PreparedChange{}, ErrInvalidChange
143+
return PreparedChange{}, fmt.Errorf("prepare queue %q: invalid boundary URI %q (length %d): %w", queue, boundary, len(boundary), ErrInvalidChange)
144144
}
145145
if current.EffectiveFromCommitURI != "" {
146146
forward, err := sc.IsAncestor(ctx, current.EffectiveFromCommitURI, boundary)
147147
if err != nil {
148148
return PreparedChange{}, err
149149
}
150150
if !forward {
151-
return PreparedChange{}, ErrInvalidChange
151+
return PreparedChange{}, fmt.Errorf("prepare queue %q: boundary %q does not include previous boundary %q: %w", queue, boundary, current.EffectiveFromCommitURI, ErrInvalidChange)
152152
}
153153
}
154154
change.next.State, change.next.EffectiveFromCommitURI = enablement, boundary
155155
change.next.ChangedAtMs = c.now().UnixMilli()
156156
if change.next.ChangedAtMs <= 0 {
157-
return PreparedChange{}, ErrInvalidChange
157+
return PreparedChange{}, fmt.Errorf("prepare queue %q: invalid enablement timestamp %d: %w", queue, change.next.ChangedAtMs, ErrInvalidChange)
158158
}
159159
return change, nil
160160
}
161161

162162
func (c *transitionController) Apply(ctx context.Context, change PreparedChange) (entity.QueueState, error) {
163163
if err := change.previous.Validate(); err != nil {
164-
return entity.QueueState{}, fmt.Errorf("%w: %v", ErrInvalidChange, err)
164+
return entity.QueueState{}, fmt.Errorf("apply queue %q: invalid prepared state version %d: %w: %v", change.previous.Queue, change.previous.Version, ErrInvalidChange, err)
165165
}
166166
stores, err := c.resolveStores(ctx, change.previous.Queue)
167167
if err != nil {
@@ -187,7 +187,7 @@ func (c *transitionController) Apply(ctx context.Context, change PreparedChange)
187187
return retained, nil
188188
}
189189
if change.previous.Version == math.MaxInt64 || internal.Version == math.MaxInt32 {
190-
return entity.QueueState{}, ErrInvalidChange
190+
return entity.QueueState{}, fmt.Errorf("apply queue %q: cannot advance enablement version %d or row version %d: %w", current.Queue, change.previous.Version, internal.Version, ErrInvalidChange)
191191
}
192192
stateVersion := change.previous.Version + 1
193193
if current.Version != change.previous.Version {
@@ -231,7 +231,7 @@ func (c *transitionController) resolveStores(ctx context.Context, queue string)
231231
return nil, err
232232
}
233233
if queue == "" || len(queue) > 255 {
234-
return nil, ErrInvalidChange
234+
return nil, fmt.Errorf("invalid queue name %q (length %d): %w", queue, len(queue), ErrInvalidChange)
235235
}
236236
stores, err := c.stores.For(storage.Config{QueueName: queue})
237237
if err != nil {
@@ -249,7 +249,7 @@ func (c *transitionController) loadInternalQueue(ctx context.Context, stores sto
249249
return entity.Queue{}, err
250250
}
251251
if queue.Name != name || queue.Version <= 0 {
252-
return entity.Queue{}, queuelog.ErrInconsistentHistory
252+
return entity.Queue{}, fmt.Errorf("load queue %q: invalid stored queue %q at row version %d: %w", name, queue.Name, queue.Version, queuelog.ErrInconsistentHistory)
253253
}
254254
if queue.EnablementVersion == 0 && queue.Enablement == entity.QueueEnablementUnknown && queue.EffectiveFromCommitURI == "" && queue.EnablementChangedAtMs == 0 {
255255
return entity.Queue{}, storage.ErrNotFound
@@ -263,7 +263,7 @@ func (c *transitionController) loadInternalQueue(ctx context.Context, stores sto
263263
func (c *transitionController) repairQueueReadModels(ctx context.Context, stores storage.Storage, queue entity.Queue) (entity.QueueState, error) {
264264
state := queue.EnablementState()
265265
if err := state.Validate(); err != nil {
266-
return entity.QueueState{}, fmt.Errorf("%w: %v", queuelog.ErrInconsistentHistory, err)
266+
return entity.QueueState{}, fmt.Errorf("queue %q: invalid internal enablement at state version %d, row version %d: %w: %v", queue.Name, state.Version, queue.Version, queuelog.ErrInconsistentHistory, err)
267267
}
268268
return state, c.materializer.PersistLog(ctx, stores, queuelog.NewQueueStateLog(queue))
269269
}
@@ -282,15 +282,15 @@ func (c *transitionController) validatePreparedBoundary(ctx context.Context, cha
282282
return err
283283
}
284284
if !belongs {
285-
return ErrInvalidChange
285+
return fmt.Errorf("queue %q: prepared boundary %q is not an ancestor of current tip %q: %w", change.previous.Queue, change.next.EffectiveFromCommitURI, tip, ErrInvalidChange)
286286
}
287287
if change.previous.EffectiveFromCommitURI != "" {
288288
forward, err := sc.IsAncestor(ctx, change.previous.EffectiveFromCommitURI, change.next.EffectiveFromCommitURI)
289289
if err != nil {
290290
return err
291291
}
292292
if !forward {
293-
return ErrInvalidChange
293+
return fmt.Errorf("queue %q: prepared boundary %q does not include previous boundary %q: %w", change.previous.Queue, change.next.EffectiveFromCommitURI, change.previous.EffectiveFromCommitURI, ErrInvalidChange)
294294
}
295295
}
296296
return nil

‎web/test/e2e/stovepipe/run_e2e.py‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,20 @@ def main():
8585
},
8686
env=env,
8787
)
88-
run([*compose, "up", "-d", "--no-build", "--pull", "never", "--wait"], env)
88+
run(
89+
[
90+
*compose,
91+
"up",
92+
"-d",
93+
"--no-build",
94+
"--pull",
95+
"never",
96+
"--wait",
97+
"mysql-app",
98+
"mysql-queue",
99+
],
100+
env,
101+
)
89102
apply_schemas(
90103
[
91104
(
@@ -117,6 +130,7 @@ def main():
117130
env,
118131
input=_pagination_fixture_sql(),
119132
)
133+
run([*compose, "up", "-d", "--no-build", "--pull", "never", "--wait"], env)
120134
for key, service, port in [
121135
("STOVEPIPE_E2E_WEB_URL", "web-service", "3000"),
122136
("STOVEPIPE_E2E_URL", "stovepipe-service", "8080"),

‎web/test/e2e/stovepipe/run_e2e_test.py‎

Lines changed: 30 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -9,9 +9,9 @@
99
from web.test.e2e.stovepipe.run_e2e import main
1010

1111

12-
class RunnerCleanupTest(unittest.TestCase):
13-
def test_termination_during_playwright_cleans_up_the_stack(self):
14-
for signum in (signal.SIGTERM, signal.SIGINT):
12+
class RunnerLifecycleTest(unittest.TestCase):
13+
def test_startup_order_and_cleanup(self):
14+
for signum in (None, signal.SIGTERM, signal.SIGINT):
1515
with self.subTest(
1616
signal=signum
1717
), tempfile.TemporaryDirectory() as directory:
@@ -25,10 +25,14 @@ def test_termination_during_playwright_cleans_up_the_stack(self):
2525
):
2626
(test_dir / name).touch()
2727
commands = []
28+
schema_application = []
29+
30+
def record_schema_application(_groups, *, env):
31+
schema_application.append(len(commands))
2832

2933
def run(command, **kwargs):
3034
commands.append(command)
31-
if command[0] == str(root / "playwright"):
35+
if command[0] == str(root / "playwright") and signum is not None:
3236
signal.raise_signal(signum)
3337
return subprocess.CompletedProcess(command, 0)
3438

@@ -58,14 +62,33 @@ def run(command, **kwargs):
5862
"subprocess.run", side_effect=run
5963
), patch(
6064
"subprocess.check_output", return_value="127.0.0.1:12345"
65+
), patch(
66+
"web.test.e2e.stovepipe.run_e2e.apply_schemas",
67+
side_effect=record_schema_application,
6168
):
62-
with self.assertRaises(SystemExit) as exit:
63-
main()
64-
self.assertEqual(exit.exception.code, 128 + signum)
69+
if signum is None:
70+
self.assertEqual(main(), 0)
71+
else:
72+
with self.assertRaises(SystemExit) as exit:
73+
main()
74+
self.assertEqual(exit.exception.code, 128 + signum)
6575
finally:
6676
for sig, handler in handlers.items():
6777
signal.signal(sig, handler)
6878

79+
startups = [
80+
(index, command)
81+
for index, command in enumerate(commands)
82+
if "up" in command
83+
]
84+
self.assertEqual(len(startups), 2)
85+
self.assertEqual(startups[0][1][-2:], ["mysql-app", "mysql-queue"])
86+
self.assertEqual(len(schema_application), 1)
87+
self.assertLess(startups[0][0], schema_application[0])
88+
self.assertLess(schema_application[0], startups[1][0])
89+
self.assertEqual(
90+
commands[schema_application[0]][:3], ["docker", "exec", "-i"]
91+
)
6992
self.assertTrue(
7093
any(
7194
command[-3:] == ["down", "-v", "--remove-orphans"]

0 commit comments

Comments
 (0)