Claim a build key by create and compare-and-set, never twice, retry a failed key except a refused kind (issue 1607, AC3)
This commit is contained in:
@@ -0,0 +1,107 @@
|
||||
package build
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// ErrConflict is what a Store answers when a Create finds the key or an Update finds another revision than the one given.
|
||||
var ErrConflict = errors.New("the key exists or changed")
|
||||
|
||||
// Store is the builder's bucket (builds): create and compare-and-set, each answering the key's new revision. Only the
|
||||
// builder's credential writes it (ADR 0429 clause 5).
|
||||
type Store interface {
|
||||
Get(ctx context.Context, key string) (value []byte, revision uint64, found bool, err error)
|
||||
Create(ctx context.Context, key string, value []byte) (uint64, error)
|
||||
Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error)
|
||||
}
|
||||
|
||||
// Claimer claims fingerprints for the builder instance By (ADR 0429 clause 5: a builder claims a fingerprint by creating
|
||||
// its key and moves it by compare-and-set).
|
||||
type Claimer struct {
|
||||
Store Store
|
||||
By string
|
||||
}
|
||||
|
||||
// Claim is a build the builder holds: its key, the revision the next write must be made at and the value last written.
|
||||
type Claim struct {
|
||||
Key string
|
||||
Rev uint64
|
||||
Build delivery.BuilderBuild
|
||||
}
|
||||
|
||||
// Claim takes the build of module at fingerprint, or answers nil when it is not this builder's to build now:
|
||||
// - no key: it is created building, by this builder; a create that finds the key made meanwhile gets nothing;
|
||||
// - a done key is never taken again (the same source is never built twice);
|
||||
// - a failed key is taken again as a new revision of the same key (ADR 0429 clause 18), except one failed for a kind no
|
||||
// builder builds yet (RefusedPrefix, ADR 0478 clause 2), which waits for another fingerprint;
|
||||
// - a key left queued or building by this builder is taken as it stands (ADR 0478 clause 5: a stopped builder takes its
|
||||
// own again when it starts); one held by another builder is left, since nothing reclaims without presence.
|
||||
func (c Claimer) Claim(ctx context.Context, module, fingerprint, manifest string) (*Claim, error) {
|
||||
key, err := delivery.BuilderBuildKey(module, fingerprint)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
building := delivery.BuilderBuild{State: delivery.BuildBuilding, By: c.By, Step: "claimed", Manifest: manifest}
|
||||
value, rev, found, err := c.Store.Get(ctx, key)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s: %w", key, err)
|
||||
}
|
||||
if !found {
|
||||
return c.write(ctx, key, building, func(data []byte) (uint64, error) { return c.Store.Create(ctx, key, data) })
|
||||
}
|
||||
cur, err := delivery.DecodeBuilderBuild(value)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s holds what is not a build: %w", key, err)
|
||||
}
|
||||
switch {
|
||||
case cur.State == delivery.BuildDone:
|
||||
return nil, nil
|
||||
case cur.State == delivery.BuildFailed && strings.HasPrefix(cur.Error, RefusedPrefix):
|
||||
return nil, nil
|
||||
case cur.State == delivery.BuildFailed:
|
||||
return c.write(ctx, key, building, func(data []byte) (uint64, error) { return c.Store.Update(ctx, key, data, rev) })
|
||||
case cur.By == c.By:
|
||||
return &Claim{Key: key, Rev: rev, Build: cur}, nil
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (c Claimer) write(_ context.Context, key string, b delivery.BuilderBuild, put func([]byte) (uint64, error)) (*Claim, error) {
|
||||
data, err := delivery.EncodeBuilderBuild(b)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rev, err := put(data)
|
||||
if errors.Is(err, ErrConflict) {
|
||||
return nil, nil // another builder made it first
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s: %w", key, err)
|
||||
}
|
||||
b.V = delivery.Version
|
||||
return &Claim{Key: key, Rev: rev, Build: b}, nil
|
||||
}
|
||||
|
||||
// Finish moves the claimed key to b by compare-and-set at the claim's revision, and writes nothing when b is the value
|
||||
// the key holds. A key that changed meanwhile is ErrConflict: the claim is no longer this builder's.
|
||||
func (c Claimer) Finish(ctx context.Context, cl *Claim, b delivery.BuilderBuild) error {
|
||||
b.V = delivery.Version
|
||||
if delivery.SameBuilderBuild(cl.Build, b) {
|
||||
return nil
|
||||
}
|
||||
data, err := delivery.EncodeBuilderBuild(b)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
rev, err := c.Store.Update(ctx, cl.Key, data, cl.Rev)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s: %w", cl.Key, err)
|
||||
}
|
||||
cl.Rev, cl.Build = rev, b
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,188 @@
|
||||
package build
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// memStore is a bucket's create and compare-and-set in memory, with a revision per write.
|
||||
type memStore struct {
|
||||
vals map[string][]byte
|
||||
revs map[string]uint64
|
||||
next uint64
|
||||
writes int
|
||||
// before runs once ahead of the next write, so a test changes the key under the writer.
|
||||
before func()
|
||||
}
|
||||
|
||||
func newMem() *memStore { return &memStore{vals: map[string][]byte{}, revs: map[string]uint64{}} }
|
||||
|
||||
func (m *memStore) Get(_ context.Context, key string) ([]byte, uint64, bool, error) {
|
||||
v, ok := m.vals[key]
|
||||
return v, m.revs[key], ok, nil
|
||||
}
|
||||
|
||||
func (m *memStore) hook() {
|
||||
if h := m.before; h != nil {
|
||||
m.before = nil
|
||||
h()
|
||||
}
|
||||
}
|
||||
|
||||
func (m *memStore) put(key string, value []byte) uint64 {
|
||||
m.next++
|
||||
m.writes++
|
||||
m.vals[key], m.revs[key] = value, m.next
|
||||
return m.next
|
||||
}
|
||||
|
||||
func (m *memStore) Create(_ context.Context, key string, value []byte) (uint64, error) {
|
||||
m.hook()
|
||||
if _, ok := m.vals[key]; ok {
|
||||
return 0, ErrConflict
|
||||
}
|
||||
return m.put(key, value), nil
|
||||
}
|
||||
|
||||
func (m *memStore) Update(_ context.Context, key string, value []byte, rev uint64) (uint64, error) {
|
||||
m.hook()
|
||||
if _, ok := m.vals[key]; !ok || m.revs[key] != rev {
|
||||
return 0, ErrConflict
|
||||
}
|
||||
return m.put(key, value), nil
|
||||
}
|
||||
|
||||
const (
|
||||
fp1 = "src2:1111111111111111111111111111111111111111111111111111111111111111"
|
||||
digest = "sha256:2222222222222222222222222222222222222222222222222222222222222222"
|
||||
)
|
||||
|
||||
func seed(t *testing.T, st *memStore, b delivery.BuilderBuild) string {
|
||||
t.Helper()
|
||||
key, err := delivery.BuilderBuildKey("m", fp1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
data, err := delivery.EncodeBuilderBuild(b)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
st.put(key, data)
|
||||
return key
|
||||
}
|
||||
|
||||
func valueOf(t *testing.T, st *memStore, key string) delivery.BuilderBuild {
|
||||
t.Helper()
|
||||
b, err := delivery.DecodeBuilderBuild(st.vals[key])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func TestClaimCreatesTheKeyBuildingAndASecondBuilderGetsNothing(t *testing.T) { // 1607/AC3
|
||||
st := newMem()
|
||||
a, b := Claimer{Store: st, By: "novox.a"}, Claimer{Store: st, By: "novox.b"}
|
||||
got, err := a.Claim(context.Background(), "m", fp1, hex64('a'))
|
||||
if err != nil || got == nil {
|
||||
t.Fatalf("first claim = %v, %v, want a claim", got, err)
|
||||
}
|
||||
if v := valueOf(t, st, got.Key); v.State != delivery.BuildBuilding || v.By != "novox.a" || v.Manifest != hex64('a') {
|
||||
t.Errorf("the key holds %+v, want building by novox.a with the manifest digest", v)
|
||||
}
|
||||
if second, err := b.Claim(context.Background(), "m", fp1, hex64('a')); err != nil || second != nil {
|
||||
t.Errorf("second builder's claim = %v, %v, want none: the same source is never built twice", second, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClaimLosingTheCreateRaceGetsNothing(t *testing.T) { // 1607/AC3
|
||||
st := newMem()
|
||||
st.before = func() { seed(t, st, delivery.BuilderBuild{State: delivery.BuildBuilding, By: "novox.b"}) }
|
||||
got, err := Claimer{Store: st, By: "novox.a"}.Claim(context.Background(), "m", fp1, "")
|
||||
if err != nil || got != nil {
|
||||
t.Errorf("claim = %v, %v, want none: the create failed because the key exists", got, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestADoneKeyIsNeverTakenAgain(t *testing.T) { // 1607/AC3
|
||||
st := newMem()
|
||||
seed(t, st, delivery.BuilderBuild{State: delivery.BuildDone, By: "novox.a", Manifest: hex64('a'), Artifacts: []delivery.BuiltArtifact{{Name: "x", Digest: digest}}})
|
||||
writes := st.writes
|
||||
got, err := Claimer{Store: st, By: "novox.a"}.Claim(context.Background(), "m", fp1, hex64('a'))
|
||||
if err != nil || got != nil || st.writes != writes {
|
||||
t.Errorf("claim = %v, %v, writes %d (was %d), want none and no write", got, err, st.writes, writes)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAFailedKeyIsTakenAgainAsANewRevisionExceptARefusedKind(t *testing.T) { // 1607/AC3, ADR 0478 clause 2
|
||||
st := newMem()
|
||||
key := seed(t, st, delivery.BuilderBuild{State: delivery.BuildFailed, By: "novox.a", Error: "the step failed"})
|
||||
rev := st.revs[key]
|
||||
got, err := Claimer{Store: st, By: "novox.a"}.Claim(context.Background(), "m", fp1, hex64('a'))
|
||||
if err != nil || got == nil || st.revs[key] <= rev {
|
||||
t.Fatalf("claim = %v, %v, revision %d (was %d), want a claim as a new revision of the same key", got, err, st.revs[key], rev)
|
||||
}
|
||||
if v := valueOf(t, st, key); v.State != delivery.BuildBuilding || v.Error != "" {
|
||||
t.Errorf("the key holds %+v, want building with no error", v)
|
||||
}
|
||||
|
||||
st = newMem()
|
||||
key = seed(t, st, delivery.BuilderBuild{State: delivery.BuildFailed, By: "novox.a", Error: RefusedPrefix + " image (artifact x): build-agent builds it"})
|
||||
writes := st.writes
|
||||
if got, err := (Claimer{Store: st, By: "novox.a"}).Claim(context.Background(), "m", fp1, ""); err != nil || got != nil || st.writes != writes {
|
||||
t.Errorf("a refused kind: claim = %v, %v, writes %d (was %d), want none and no write", got, err, st.writes, writes)
|
||||
}
|
||||
}
|
||||
|
||||
func TestABuildLeftBuildingBySameBuilderIsResumedAndAnotherBuildersIsLeft(t *testing.T) { // 1607/AC3, ADR 0478 clause 5
|
||||
st := newMem()
|
||||
seed(t, st, delivery.BuilderBuild{State: delivery.BuildBuilding, By: "novox.a"})
|
||||
if got, err := (Claimer{Store: st, By: "novox.a"}).Claim(context.Background(), "m", fp1, ""); err != nil || got == nil {
|
||||
t.Errorf("own building key: claim = %v, %v, want it taken again", got, err)
|
||||
}
|
||||
if got, err := (Claimer{Store: st, By: "novox.b"}).Claim(context.Background(), "m", fp1, ""); err != nil || got != nil {
|
||||
t.Errorf("another builder's building key: claim = %v, %v, want none: nothing reclaims without presence", got, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFinishMovesTheKeyByCompareAndSetAndWritesOnlyADifferentValue(t *testing.T) { // 1607/AC3
|
||||
st := newMem()
|
||||
c := Claimer{Store: st, By: "novox.a"}
|
||||
got, _ := c.Claim(context.Background(), "m", fp1, hex64('a'))
|
||||
done := delivery.BuilderBuild{State: delivery.BuildDone, By: "novox.a", Manifest: hex64('a'), Artifacts: []delivery.BuiltArtifact{{Name: "x", Kind: "archive", Digest: digest}}}
|
||||
if err := c.Finish(context.Background(), got, done); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if v := valueOf(t, st, got.Key); v.State != delivery.BuildDone || v.Artifacts[0].Digest != digest {
|
||||
t.Errorf("the key holds %+v", v)
|
||||
}
|
||||
writes := st.writes
|
||||
if err := c.Finish(context.Background(), got, done); err != nil && !errors.Is(err, ErrConflict) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if st.writes != writes {
|
||||
t.Errorf("the same value was written again (%d writes, was %d)", st.writes, writes)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFinishAfterTheKeyChangedIsAConflictNotAnOverwrite(t *testing.T) { // 1607/AC3
|
||||
st := newMem()
|
||||
c := Claimer{Store: st, By: "novox.a"}
|
||||
got, _ := c.Claim(context.Background(), "m", fp1, "")
|
||||
st.put(got.Key, []byte(`{"v":1,"state":"failed","error":"someone else"}`))
|
||||
err := c.Finish(context.Background(), got, delivery.BuilderBuild{State: delivery.BuildFailed, Error: "mine"})
|
||||
if !errors.Is(err, ErrConflict) {
|
||||
t.Errorf("Finish = %v, want ErrConflict", err)
|
||||
}
|
||||
}
|
||||
|
||||
func hex64(c byte) string {
|
||||
b := make([]byte, 64)
|
||||
for i := range b {
|
||||
b[i] = c
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
Reference in New Issue
Block a user