19d4f5e fix(db): make failure win over close in Apply and group tests deterministic · ermite-kv

M db.go +20 -8


@@ -217,12 +217,8 @@ func (db DB) Apply(ctx context.Context, b Batch) (uint64, error) {
217	217		if b.encodedSize() > db.opts.MaxBatchBytes {
218	218		return 0, ErrBatchTooLarge
219	219		}
220		-	select {
221		-	case <-db.closing:
222		-	return 0, ErrClosed
223		-	case <-db.failed:
224		-	return 0, db.failErr
225		-	default:
	220	+	if err := db.stopped(); err != nil {
	221	+	return 0, err
226	222		}
227	223		if err := ctx.Err(); err != nil {
228	224		return 0, err

@@ -235,11 +231,27 @@ func (db DB) Apply(ctx context.Context, b Batch) (uint64, error) {
235	231		case <-ctx.Done():
236	232		return 0, r.release(ctx.Err())
237	233		case <-db.closing:
238		-	return 0, r.release(ErrClosed)
	234	+	return 0, r.release(db.stopped())
239	235		case <-db.failed:
240		-	return 0, r.release(db.failErr)
	236	+	return 0, r.release(db.stopped())
241	237		}
242	238		res := <-r.reply // the writer always answers a request it has taken
243	239		_ = r.release(nil)
244	240		return res.seq, res.err
245	241		}
	242	+
	243	+	// stopped returns the failure (which wins over Close, so callers see a
	244	+	// stable error), ErrClosed, or nil while the database accepts batches.
	245	+	func (db *DB) stopped() error {
	246	+	select {
	247	+	case <-db.failed:
	248	+	return db.failErr
	249	+	default:
	250	+	}
	251	+	select {
	252	+	case <-db.closing:
	253	+	return ErrClosed
	254	+	default:
	255	+	return nil
	256	+	}
	257	+	}

 	if b.encodedSize() > db.opts.MaxBatchBytes {
 		return 0, ErrBatchTooLarge
 	}
-	select {
+	if err := db.stopped(); err != nil {
-	case <-db.closing:
+		return 0, err
-		return 0, ErrClosed
-	case <-db.failed:
-		return 0, db.failErr
-	default:
 	}
 	if err := ctx.Err(); err != nil {
 		return 0, err
 	case <-ctx.Done():
 		return 0, r.release(ctx.Err())
 	case <-db.closing:
-		return 0, r.release(ErrClosed)
+		return 0, r.release(db.stopped())
 	case <-db.failed:
-		return 0, r.release(db.failErr)
+		return 0, r.release(db.stopped())
 	}
 	res := <-r.reply // the writer always answers a request it has taken
 	_ = r.release(nil)
 	return res.seq, res.err
 }
+// stopped returns the failure (which wins over Close, so callers see a
+// stable error), ErrClosed, or nil while the database accepts batches.
+func (db *DB) stopped() error {
+	select {
+	case <-db.failed:
+		return db.failErr
+	default:
+	}
+	select {
+	case <-db.closing:
+		return ErrClosed
+	default:
+		return nil
+	}
+}

M db_test.go +45 -16


@@ -351,30 +351,57 @@ func TestPreconditionsBeforeOps(t *testing.T) {
351	351		mustMissing(t, db.Snapshot(bg), "other")
352	352		}
353	353
	354	+	// holdWriter applies a batch whose sync is slow and returns once the writer
	355	+	// is inside that sync, so batches applied next queue up and form one group
	356	+	// (GroupWindow 0 drains everything already queued). wait returns when the
	357	+	// held batch is acknowledged; the sync delay stays set until then.
	358	+	func holdWriter(t testing.T, db DB, fx *faultfs.FS) (wait func()) {
	359	+	t.Helper()
	360	+	base := fx.Syncs()
	361	+	fx.SetSyncDelay(300 * time.Millisecond)
	362	+	done := make(chan error, 1)
	363	+	go func() {
	364	+	var b Batch
	365	+	b.Put([]byte("hold"), nil)
	366	+	_, err := db.Apply(bg, &b)
	367	+	done <- err
	368	+	}()
	369	+	for fx.Syncs() == base {
	370	+	time.Sleep(time.Millisecond)
	371	+	}
	372	+	return func() {
	373	+	t.Helper()
	374	+	if err := <-done; err != nil {
	375	+	t.Fatalf("held Apply: %v", err)
	376	+	}
	377	+	fx.SetSyncDelay(0)
	378	+	}
	379	+	}
	380	+
354	381		// TestGroupPreconditions checks that preconditions see earlier batches of
355		-	// the same group: with a group window, the second batch must conflict.
	382	+	// the same group: the second batch of the group must conflict.
356	383		func TestGroupPreconditions(t *testing.T) {
357	384		fx := faultfs.New()
358		-	db := mustOpen(t, fx, Options{GroupWindow: 200 * time.Millisecond})
	385	+	db := mustOpen(t, fx, Options{})
359	386		syncs := fx.Syncs()
	387	+	wait := holdWriter(t, db, fx)
360	388		errs := make([]error, 2)
361	389		var wg sync.WaitGroup
362	390		for i := range errs {
363		-	wg.Add(1)
364		-	go func() {
365		-	defer wg.Done()
	391	+	wg.Go(func() {
366	392		var b Batch
367	393		b.RequireAbsent([]byte("k"))
368	394		b.Put([]byte("k"), []byte{byte('a' + i)})
369	395		_, errs[i] = db.Apply(bg, &b)
370		-	}()
	396	+	})
371	397		}
372	398		wg.Wait()
	399	+	wait()
373	400		if (errs[0] == nil) == (errs[1] == nil) {
374	401		t.Fatalf("errors %v, want exactly one success", errs)
375	402		}
376		-	if fx.Syncs() != syncs+1 {
377		-	t.Fatalf("%d syncs for the group, want 1", fx.Syncs()-syncs)
	403	+	if got := fx.Syncs() - syncs; got != 2 {
	404	+	t.Fatalf("%d syncs, want 2 (held batch, then one group)", got)
378	405		}
379	406		}
380	407

@@ -672,7 +699,7 @@ func TestContextBeforeDequeue(t *testing.T) {
672	699		_, err := db.Apply(ctx, &b)
673	700		second <- err
674	701		}()
675		-	time.Sleep(10 * time.Millisecond)
	702	+	time.Sleep(50 * time.Millisecond) // let Apply reach its blocking select
676	703		cancel()
677	704		if err := <-second; !errors.Is(err, context.Canceled) {
678	705		t.Fatalf("Apply = %v, want context.Canceled", err)

@@ -787,11 +814,11 @@ func testInjectedFailure(t *testing.T, f failure, powerLoss bool) {
787	814		fx := faultfs.New()
788	815		var logs syncBuffer
789	816		ctx := zerowrap.WithCtx(bg, zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logs}))
790		-	db := mustOpenCtx(t, ctx, fx, Options{GroupWindow: 150 * time.Millisecond})
791		-	putKV(t, db, "before", "1")
	817	+	db := mustOpenCtx(t, ctx, fx, Options{})
	818	+	wait := holdWriter(t, db, fx) // seq 1: "hold"; its write and sync have started
792	819		f.inject(fx)
793	820
794		-	// Three batches form the failed group (the window gathers them).
	821	+	// Three batches queue behind the held one and form the failed group.
795	822		const group = 3
796	823		errs := make([]error, group)
797	824		var wg sync.WaitGroup

@@ -803,6 +830,7 @@ func testInjectedFailure(t *testing.T, f failure, powerLoss bool) {
803	830		})
804	831		}
805	832		wg.Wait()
	833	+	wait()
806	834		for i, err := range errs {
807	835		if !errors.Is(err, ErrFailed) {
808	836		t.Fatalf("batch %d: %v, want ErrFailed", i, err)

@@ -831,7 +859,7 @@ func testInjectedFailure(t *testing.T, f failure, powerLoss bool) {
831	859		if s.Seq() != 1 \|\| s.Len() != 1 {
832	860		t.Fatalf("snapshot after failure: seq %d len %d", s.Seq(), s.Len())
833	861		}
834		-	mustGet(t, s, "before", "1")
	862	+	mustGet(t, s, "hold", "")
835	863		mustMissing(t, s, "failed0")
836	864		if err := db.Close(bg); !errors.Is(err, ErrFailed) {
837	865		t.Fatalf("Close = %v, want the failure", err)

@@ -844,7 +872,7 @@ func testInjectedFailure(t *testing.T, f failure, powerLoss bool) {
844	872		fx.PowerLoss(faultfs.PowerLossOptions{})
845	873		}
846	874		s = mustOpen(t, fx, Options{}).Snapshot(bg)
847		-	mustGet(t, s, "before", "1")
	875	+	mustGet(t, s, "hold", "")
848	876		present := 0
849	877		for i := range group {
850	878		if _, ok := s.Get(bg, fmt.Appendf(nil, "failed%d", i)); ok {

@@ -904,7 +932,8 @@ func (s *syncBuffer) String() string {
904	932		return s.b.String()
905	933		}
906	934
907		-	// powerLossBase returns the first seed; ERMITE_KV_SEED reproduces a run.
	935	+	// powerLossBase returns the first seed. ERMITE_KV_SEED replays the same
	936	+	// parameters; goroutine scheduling still varies between runs.
908	937		func powerLossBase(t *testing.T) uint64 {
909	938		if v := os.Getenv("ERMITE_KV_SEED"); v != "" {
910	939		seed, err := strconv.ParseUint(v, 10, 64)

@@ -946,7 +975,7 @@ func TestPowerLoss(t *testing.T) {
946	975		for i := range iterations {
947	976		seed := base + uint64(i)
948	977		if !runPowerLoss(t, seed, m.opts) {
949		-	t.Fatalf("failed with seed %d (ERMITE_KV_SEED=%d reproduces it)", seed, seed)
	978	+	t.Fatalf("failed with seed %d (ERMITE_KV_SEED=%d replays its parameters)", seed, seed)
950	979		}
951	980		}
952	981		})

 	mustMissing(t, db.Snapshot(bg), "other")
 }
+// holdWriter applies a batch whose sync is slow and returns once the writer
+// is inside that sync, so batches applied next queue up and form one group
+// (GroupWindow 0 drains everything already queued). wait returns when the
+// held batch is acknowledged; the sync delay stays set until then.
+func holdWriter(t *testing.T, db *DB, fx *faultfs.FS) (wait func()) {
+	t.Helper()
+	base := fx.Syncs()
+	fx.SetSyncDelay(300 * time.Millisecond)
+	done := make(chan error, 1)
+	go func() {
+		var b Batch
+		b.Put([]byte("hold"), nil)
+		_, err := db.Apply(bg, &b)
+		done <- err
+	}()
+	for fx.Syncs() == base {
+		time.Sleep(time.Millisecond)
+	}
+	return func() {
+		t.Helper()
+		if err := <-done; err != nil {
+			t.Fatalf("held Apply: %v", err)
+		}
+		fx.SetSyncDelay(0)
+	}
+}
 // TestGroupPreconditions checks that preconditions see earlier batches of
-// the same group: with a group window, the second batch must conflict.
+// the same group: the second batch of the group must conflict.
 func TestGroupPreconditions(t *testing.T) {
 	fx := faultfs.New()
-	db := mustOpen(t, fx, Options{GroupWindow: 200 * time.Millisecond})
+	db := mustOpen(t, fx, Options{})
 	syncs := fx.Syncs()
+	wait := holdWriter(t, db, fx)
 	errs := make([]error, 2)
 	var wg sync.WaitGroup
 	for i := range errs {
-		wg.Add(1)
+		wg.Go(func() {
-		go func() {
-			defer wg.Done()
 			var b Batch
 			b.RequireAbsent([]byte("k"))
 			b.Put([]byte("k"), []byte{byte('a' + i)})
 			_, errs[i] = db.Apply(bg, &b)
-		}()
+		})
 	}
 	wg.Wait()
+	wait()
 	if (errs[0] == nil) == (errs[1] == nil) {
 		t.Fatalf("errors %v, want exactly one success", errs)
 	}
-	if fx.Syncs() != syncs+1 {
+	if got := fx.Syncs() - syncs; got != 2 {
-		t.Fatalf("%d syncs for the group, want 1", fx.Syncs()-syncs)
+		t.Fatalf("%d syncs, want 2 (held batch, then one group)", got)
 	}
 }
 		_, err := db.Apply(ctx, &b)
 		second <- err
 	}()
-	time.Sleep(10 * time.Millisecond)
+	time.Sleep(50 * time.Millisecond) // let Apply reach its blocking select
 	cancel()
 	if err := <-second; !errors.Is(err, context.Canceled) {
 		t.Fatalf("Apply = %v, want context.Canceled", err)
 	fx := faultfs.New()
 	var logs syncBuffer
 	ctx := zerowrap.WithCtx(bg, zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logs}))
-	db := mustOpenCtx(t, ctx, fx, Options{GroupWindow: 150 * time.Millisecond})
+	db := mustOpenCtx(t, ctx, fx, Options{})
-	putKV(t, db, "before", "1")
+	wait := holdWriter(t, db, fx) // seq 1: "hold"; its write and sync have started
 	f.inject(fx)
-	// Three batches form the failed group (the window gathers them).
+	// Three batches queue behind the held one and form the failed group.
 	const group = 3
 	errs := make([]error, group)
 	var wg sync.WaitGroup
 		})
 	}
 	wg.Wait()
+	wait()
 	for i, err := range errs {
 		if !errors.Is(err, ErrFailed) {
 			t.Fatalf("batch %d: %v, want ErrFailed", i, err)
 	if s.Seq() != 1 || s.Len() != 1 {
 		t.Fatalf("snapshot after failure: seq %d len %d", s.Seq(), s.Len())
 	}
-	mustGet(t, s, "before", "1")
+	mustGet(t, s, "hold", "")
 	mustMissing(t, s, "failed0")
 	if err := db.Close(bg); !errors.Is(err, ErrFailed) {
 		t.Fatalf("Close = %v, want the failure", err)
 		fx.PowerLoss(faultfs.PowerLossOptions{})
 	}
 	s = mustOpen(t, fx, Options{}).Snapshot(bg)
-	mustGet(t, s, "before", "1")
+	mustGet(t, s, "hold", "")
 	present := 0
 	for i := range group {
 		if _, ok := s.Get(bg, fmt.Appendf(nil, "failed%d", i)); ok {
 	return s.b.String()
 }
-// powerLossBase returns the first seed; ERMITE_KV_SEED reproduces a run.
+// powerLossBase returns the first seed. ERMITE_KV_SEED replays the same
+// parameters; goroutine scheduling still varies between runs.
 func powerLossBase(t *testing.T) uint64 {
 	if v := os.Getenv("ERMITE_KV_SEED"); v != "" {
 		seed, err := strconv.ParseUint(v, 10, 64)
 			for i := range iterations {
 				seed := base + uint64(i)
 				if !runPowerLoss(t, seed, m.opts) {
-					t.Fatalf("failed with seed %d (ERMITE_KV_SEED=%d reproduces it)", seed, seed)
+					t.Fatalf("failed with seed %d (ERMITE_KV_SEED=%d replays its parameters)", seed, seed)
 				}
 			}
 		})

M writer.go +3 -5


@@ -45,11 +45,9 @@ func (db *DB) writer() {
45	45		defer func() {
46	46		// A request taken but never grouped must still be answered.
47	47		if pending != nil {
48		-	err := ErrClosed
49		-	select {
50		-	case <-db.failed:
51		-	err = db.failErr
52		-	default:
	48	+	err := db.stopped()
	49	+	if err == nil {
	50	+	err = ErrClosed // not reachable: the writer only exits on close or failure
53	51		}
54	52		pending.reply <- result{err: err}
55	53		}

 	defer func() {
 		// A request taken but never grouped must still be answered.
 		if pending != nil {
-			err := ErrClosed
+			err := db.stopped()
-			select {
+			if err == nil {
-			case <-db.failed:
+				err = ErrClosed // not reachable: the writer only exits on close or failure
-				err = db.failErr
-			default:
 			}
 			pending.reply <- result{err: err}
 		}