The indexer binds its taken state when it appears, as issue 1790 asks of every state a delivery adds: an indexer started before the bus holds it still takes merges in and says the key it did not write (issue 1280, ADR 0556 clause 1)
This commit is contained in:
@@ -305,10 +305,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
taken, err := process.EssentialBucket(ctx, js, "indexer", delivery.TakenBucket, "say the merge a repository's modules were taken in at")
|
// Its taken state (ADR 0556 clause 1) is bound when it appears, as every state a delivery adds (issue 1790): until then
|
||||||
if err != nil {
|
// a merge's taken key is said and not written, and the rest of the indexer runs.
|
||||||
return err
|
taken := process.AwaitBucket(ctx, js, conn, delivery.TakenBucket, "no repository's taken key is written", logger.Printf)
|
||||||
}
|
|
||||||
// Its placement state (ADR 0549 §2) is bound when it appears, as every state a delivery adds (issue 1790): until then a
|
// Its placement state (ADR 0549 §2) is bound when it appears, as every state a delivery adds (issue 1790): until then a
|
||||||
// catalogue merge's placement is said and not written, and the rest of the indexer runs.
|
// catalogue merge's placement is said and not written, and the rest of the indexer runs.
|
||||||
placement := process.AwaitBucket(ctx, js, conn, delivery.PlacementBucket, "no catalogue merge's placement is written", logger.Printf)
|
placement := process.AwaitBucket(ctx, js, conn, delivery.PlacementBucket, "no catalogue merge's placement is written", logger.Printf)
|
||||||
@@ -345,7 +344,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
defer calls.Close()
|
defer calls.Close()
|
||||||
ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: kvStore{taken}, Builds: known, Logf: logger.Printf,
|
ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, Logf: logger.Printf,
|
||||||
Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call),
|
Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call),
|
||||||
Placement: awaitedStore{placement}, Say: func(event string, body any) error { return calls.Emit(event, body) }}
|
Placement: awaitedStore{placement}, Say: func(event string, body any) error { return calls.Emit(event, body) }}
|
||||||
// The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 §1): its key
|
// The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 §1): its key
|
||||||
|
|||||||
@@ -17,8 +17,9 @@ const (
|
|||||||
|
|
||||||
// memStore is the candidates bucket's compare-and-set behaviour in memory: a revision per write, a conflict when the
|
// memStore is the candidates bucket's compare-and-set behaviour in memory: a revision per write, a conflict when the
|
||||||
// revision read is no longer the key's. Writes counts the writes that went through; beforeWrite lets a test change the
|
// revision read is no longer the key's. Writes counts the writes that went through; beforeWrite lets a test change the
|
||||||
// key under the indexer.
|
// key under the indexer; unbound, when set, is every read's error (a state not yet on the bus).
|
||||||
type memStore struct {
|
type memStore struct {
|
||||||
|
unbound error
|
||||||
vals map[string][]byte
|
vals map[string][]byte
|
||||||
revs map[string]uint64
|
revs map[string]uint64
|
||||||
next uint64
|
next uint64
|
||||||
@@ -29,6 +30,9 @@ type memStore struct {
|
|||||||
func newMem() *memStore { return &memStore{vals: map[string][]byte{}, revs: map[string]uint64{}} }
|
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) {
|
func (m *memStore) Get(_ context.Context, key string) ([]byte, uint64, bool, error) {
|
||||||
|
if m.unbound != nil {
|
||||||
|
return nil, 0, false, m.unbound
|
||||||
|
}
|
||||||
v, ok := m.vals[key]
|
v, ok := m.vals[key]
|
||||||
return v, m.revs[key], ok, nil
|
return v, m.revs[key], ok, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
|
"git.novox.be/novox/mesh-sdk/go/process"
|
||||||
)
|
)
|
||||||
|
|
||||||
const merge = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
const merge = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
||||||
@@ -111,6 +112,13 @@ func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value
|
|||||||
m.Merge = next
|
m.Merge = next
|
||||||
return ix.Apply(context.Background(), m)
|
return ix.Apply(context.Background(), m)
|
||||||
}},
|
}},
|
||||||
|
{name: "a merge while the taken state is not on the bus yet is still taken in", writes: 1, commit: map[string]string{"a": next, "b": merge, "c": merge}, taken: merge,
|
||||||
|
then: func(ix *Indexer) error { // issue 1790: the state is bound when it appears, so the indexer runs before it does
|
||||||
|
ix.Taken = &memStore{unbound: process.ErrNotYet}
|
||||||
|
m := mergedMsg(repo, map[string]string{"modules/a": "treeA2", "modules/b": "treeB", "modules/c": "treeC"})
|
||||||
|
m.Merge = next
|
||||||
|
return ix.Apply(context.Background(), m)
|
||||||
|
}},
|
||||||
{name: "an earlier merge heard again after a later one leaves taken", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge,
|
{name: "an earlier merge heard again after a later one leaves taken", writes: 0, commit: map[string]string{"a": merge, "b": merge, "c": merge}, taken: merge,
|
||||||
then: func(ix *Indexer) error {
|
then: func(ix *Indexer) error {
|
||||||
ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { return commit == next && in == merge, nil }
|
ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) { return commit == next && in == merge, nil }
|
||||||
|
|||||||
@@ -2,9 +2,11 @@ package index
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"errors"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
|
"git.novox.be/novox/mesh-sdk/go/process"
|
||||||
)
|
)
|
||||||
|
|
||||||
// What the indexer took in (novox/hq ADR 0556 clause 1, issue 1280): after a merged message is applied whole, the key of
|
// What the indexer took in (novox/hq ADR 0556 clause 1, issue 1280): after a merged message is applied whole, the key of
|
||||||
@@ -23,7 +25,7 @@ func (ix *Indexer) writeTaken(ctx context.Context, f delivery.Forge) error {
|
|||||||
ix.logf("%s at %s: no taken key is written: %v", f.Repository, f.Merge, err)
|
ix.logf("%s at %s: no taken key is written: %v", f.Repository, f.Merge, err)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
return write(ctx, ix.Taken, key, func(value []byte, found bool) ([]byte, bool, error) {
|
err = write(ctx, ix.Taken, key, func(value []byte, found bool) ([]byte, bool, error) {
|
||||||
if found {
|
if found {
|
||||||
if cur, err := delivery.DecodeTaken(value); err == nil {
|
if cur, err := delivery.DecodeTaken(value); err == nil {
|
||||||
if cur.Merge == f.Merge {
|
if cur.Merge == f.Merge {
|
||||||
@@ -48,4 +50,11 @@ func (ix *Indexer) writeTaken(ctx context.Context, f delivery.Forge) error {
|
|||||||
data, err := delivery.EncodeTaken(delivery.Taken{Repository: delivery.RepositoryName(f.Repository), Merge: f.Merge, At: now().UTC()})
|
data, err := delivery.EncodeTaken(delivery.Taken{Repository: delivery.RepositoryName(f.Repository), Merge: f.Merge, At: now().UTC()})
|
||||||
return data, false, err
|
return data, false, err
|
||||||
})
|
})
|
||||||
|
if errors.Is(err, process.ErrNotYet) {
|
||||||
|
// The state is not on the bus yet (a delivery that adds it, bound when it appears as issue 1790 says): said, and
|
||||||
|
// the merge's modules are still taken in; the next merge of the repository writes its key once the state appears.
|
||||||
|
ix.logf("%s at %s: its taken key is not written: %v", f.Repository, f.Merge, err)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user