Prove the assigner replaced mid-pass resumes with nothing lost or written again (issue 1906 D8)

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go

@@ -5,6 +5,7 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"io"
 	"os"
 	"path/filepath"
@@ -84,6 +85,14 @@ func startBed(t *testing.T, leaveOut string) *bed {
 // start runs the program as the node does, through run with its environment and a broker file; it answers the
 // channel run's error arrives on.
 func (b *bed) start(t *testing.T) <-chan error {
+	t.Helper()
+	done, _ := b.startStoppable(t)
+	return done
+}
+
+// startStoppable is start, and answers the function that stops this one program (a replaced assigner is stopped, then
+// another started).
+func (b *bed) startStoppable(t *testing.T) (<-chan error, func()) {
 	t.Helper()
 	broker := filepath.Join(b.dir, "broker")
 	cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-assigner", User: "n1.mesh-delivery-assigner", Password: "pw"})
@@ -94,14 +103,15 @@ func (b *bed) start(t *testing.T) <-chan error {
 	ctx, stop := context.WithCancel(b.ctx)
 	done := make(chan error, 1)
 	go func() { done <- run(ctx, func(k string) string { return env[k] }, io.Discard) }()
-	t.Cleanup(func() {
+	stopped := func() {
 		stop()
 		select {
 		case <-done:
 		case <-time.After(10 * time.Second):
 		}
-	})
-	return done
+	}
+	t.Cleanup(stopped)
+	return done, stopped
 }
 
 // 1297/B2b: started whole on its credential, the assigner says on the bus the four states it keeps, each readable,
@@ -374,3 +384,58 @@ func TestAFailedDataModuleIsAskedToRestore(t *testing.T) {
 		t.Fatal("no condition for the held fingerprint was said")
 	}
 }
+
+// 1906/D8 (scenario D of acceptance test 18, ADR 0539 §19), whole on the bus: the assigner is stopped while it writes the wants
+// of forty modules and another is started; every module's want is written, and each exactly once, by revision: nothing lost
+// and nothing written again.
+func TestTheAssignerReplacedResumesWithNothingLostOrRepeated(t *testing.T) {
+	b := startBed(t, "")
+	const modules = 40
+	on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
+	b.putJSON(t, delivery.PresenceBucket, "ace", on)
+	names := make([]string, modules)
+	for i := range names {
+		name := fmt.Sprintf("mod%02d", i)
+		names[i] = name
+		fp := delivery.SourcePrefix2 + strings.Repeat(string(rune('a'+i%6)), 63) + fmt.Sprintf("%x", i%16)
+		digest := b.manifest(t, name, fmt.Sprintf(`{"module":%q,"version":"1"}`, name))
+		mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/" + name, Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
+		b.putJSON(t, delivery.ModulesBucket, name, mod)
+		built, err := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: digest})
+		if err != nil {
+			t.Fatal(err)
+		}
+		bkey, _ := delivery.BuilderBuildKey(name, fp)
+		b.putJSON(t, delivery.BuilderBuildsBucket, bkey, built)
+		gkey, _ := delivery.ControllerGrantKey(name, fp)
+		grant, _ := delivery.EncodeControllerGrant(delivery.ControllerGrant{})
+		b.putJSON(t, delivery.ControllerGrantsBucket, gkey, grant)
+		b.putJSON(t, provider.AssignmentsBucket, "ace."+name, []byte(fmt.Sprintf(`{"node":"ace","module":%q,"build":"b-%d"}`, name, i)))
+	}
+	count := func() int {
+		keys, err := b.buckets[delivery.AssignerWantsBucket].Keys(b.ctx)
+		if err != nil {
+			return 0
+		}
+		return len(keys)
+	}
+	_, stop := b.startStoppable(t)
+	for deadline := time.Now().Add(20 * time.Second); count() == 0 && time.Now().Before(deadline); time.Sleep(time.Millisecond) {
+	}
+	stop() // replaced, wherever its pass had got to
+	b.start(t)
+	for deadline := time.Now().Add(30 * time.Second); count() < modules && time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
+	}
+	time.Sleep(time.Second) // and no pass of the second writes any again
+	for _, name := range names {
+		key, _ := delivery.WantKey("ace", name)
+		history, err := b.buckets[delivery.AssignerWantsBucket].History(b.ctx, key)
+		if err != nil {
+			t.Errorf("the want of %s was never written: %v", name, err)
+			continue
+		}
+		if len(history) != 1 {
+			t.Errorf("the want of %s was written %d times, want once", name, len(history))
+		}
+	}
+}