Drop the plan state from the store and the assignments: migration 0108 and placement-only assignments (release 6, issue 1945 B9, B4)

This commit is contained in:
2026-10-24 16:42:02 +02:00
parent 96d7784bce
commit 788fb681ec
110 changed files with 512 additions and 10584 deletions
@@ -243,54 +243,6 @@ func TestObjectsAreInTheBucketBeforeTheDeclarationIsSent(t *testing.T) { // 1696
}
}
// fallbackDelivery refuses the first declaration it is sent with the bucket's failure, as a bus that has not yet been given the
// controller's grant does, and takes the second.
type fallbackDelivery struct {
bodies [][]byte
first error
}
func (f *fallbackDelivery) grant(context.Context, []readyNode) error { return nil }
func (f *fallbackDelivery) declare(_ context.Context, _ readyNode, body []byte) (string, error) {
f.bodies = append(f.bodies, body)
if len(f.bodies) == 1 && f.first != nil {
return "", fmt.Errorf("x: %w: %w", errFilesNotPut, f.first)
}
return "digest", nil
}
// 1696/AC5: when the bucket cannot be written yet (the grant travels in the declaration of the machine holding the bus), that send goes
// with every content inline and is made; a send that fails for another reason is not retried inline.
func TestASendWhoseFilesCannotBePutGoesInlineOnce(t *testing.T) { // 1696/AC5
users := contentOf("users", 100<<10)
s := readyNode{node: "novox", declared: sendable{fileReferences: true, Resources: []map[string]any{aFileResource("nats.bus-users", users)}}}
f := &fallbackDelivery{first: errors.New("timeout")}
digests, err := deliver(t.Context(), f, "novox", []readyNode{s})
if err != nil || digests["novox"] != "digest" || len(f.bodies) != 2 {
t.Fatalf("%v %v %d", digests, err, len(f.bodies))
}
if strings.Contains(string(f.bodies[0]), users[:100]) || !strings.Contains(string(f.bodies[1]), users[:100]) {
t.Error("the first body should name the file by digest and the second carry it inline")
}
other := &fallbackDelivery{}
other.first = nil
failing := &failingDelivery{err: errors.New("the bus is down")}
if _, err := deliver(t.Context(), failing, "novox", []readyNode{s}); err == nil || failing.calls != 1 {
t.Fatalf("a failure of another kind: %v after %d send(s)", err, failing.calls)
}
}
type failingDelivery struct {
err error
calls int
}
func (f *failingDelivery) grant(context.Context, []readyNode) error { return nil }
func (f *failingDelivery) declare(context.Context, readyNode, []byte) (string, error) {
f.calls++
return "", f.err
}
// A bucket for the sweep.
type sweepBucketFake struct {
objects map[string]time.Time // zero: named at the last look
@@ -502,33 +454,3 @@ func whileHeld(node string) bool {
return true
}
}
// 1696/AC9: a send holds the node's lock from the put to the publication, so the sweep never removes an object between them. The
// test goes through overTheBus.declare: while another holds the lock the declaration is not published, and at the publication the
// lock is held.
func TestASendHoldsTheNodesLockFromThePutToThePublication(t *testing.T) { // 1696/AC9
const node = "lock-send-node"
bus := &aBusThatSays{published: make(chan string, 1)}
bus.heldAtPublish = func() bool { return whileHeld(node) }
send := overTheBus{server: link.SendingOn(bus), signer: aSignerOfZeros{}}
release := lockFilesOf(node) // a sweep is judging this node's bucket
done := make(chan error, 1)
go func() {
_, err := send.declare(t.Context(), readyNode{node: node}, []byte(`{}`))
done <- err
}()
select {
case n := <-bus.published:
t.Fatalf("%s was sent its declaration while the sweep held the node's bucket", n)
case <-time.After(150 * time.Millisecond):
}
release()
select {
case <-bus.published:
case <-time.After(5 * time.Second):
t.Fatal("the send did not go on once the sweep let go")
}
if err := <-done; err == nil || !strings.Contains(err.Error(), "nothing is recorded") {
t.Fatalf("the send did not reach the publication (a lock held at it): %v", err)
}
}