diff --git a/modules/gitea/README.md b/modules/gitea/README.md index 5b0f8934..75221eb7 100644 --- a/modules/gitea/README.md +++ b/modules/gitea/README.md @@ -208,6 +208,8 @@ deletes the versions nothing keeps. | `npm_retention` | a | the versions retention would delete, and the bytes. **A dry run unless `dry_run` is false**; a real run needs `why` | | `go_packages` | r | every Go module of the registry's owner (the Go seat's), each version newest first with its publish time, its size and why it is kept: a `go.mod` on a default branch requires it, or it is among the newest `keep` | | `go_retention` | a | the Go module versions retention would delete, and the bytes; as `npm_retention`, and a `go.mod` that could not be read stops a real run. Each deletion says `removed` on the Go seat | +| `npm_incoming` | a | the npm seat's open transfers in the module's object store `incoming`, its fill and the cap its binding carries; with `drop` and `why`, drops one by hand. A transfer with no chunk for 24 hours is dropped on its own, when its key in the state `idle` ends | +| `go_incoming` | a | the same for the Go seat | **A version is kept when:** diff --git a/modules/gitea/cmd/package-registry/incoming.go b/modules/gitea/cmd/package-registry/incoming.go new file mode 100644 index 00000000..eda80b2a --- /dev/null +++ b/modules/gitea/cmd/package-registry/incoming.go @@ -0,0 +1,105 @@ +package main + +// The package seats' incoming, read and steered (design 51 *Publishing*; tools review F4): npm_incoming and +// go_incoming list a seat's open transfers, incoming's fill and the cap its binding carries, and drop one by hand. +// The holder counts against that cap, never a number of its own, and drops a transfer left with no chunk for 24 hours +// when its idle key ends (the bus's marker), never on a timer. + +import ( + "errors" + "fmt" + "os" + "strings" + "sync/atomic" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/busstate" + + "gitea/internal/packageseats" +) + +// incomingTool is a seat's _incoming. +func incomingTool(typ, seat string, holder func() (*packageseats.Holder, error)) stdio.Tool { + str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } + return stdio.Tool{ + Name: typ + "_incoming", + Description: "The " + seat + " seat's open transfers in the forge module's object store incoming (each with its " + + "opener, its parts and when it last took a chunk), incoming's fill and the cap its binding carries; a " + + "transfer with no chunk for 24 hours is dropped on its own. With drop, drops one by hand, its chunks with " + + "it (a hand act, why required). (a)", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "drop": map[string]any{"type": "string", "description": "a transfer id (the caller's own name) to drop"}, + "why": map[string]any{"type": "string", "description": "why it is dropped"}, + }}, + Run: func(args map[string]any) (any, error) { + h, err := holder() + if err != nil { + return nil, err + } + if id := str(args, "drop"); id != "" { + if str(args, "why") == "" { + return nil, errors.New("a drop is a hand act: say why") + } + t, err := h.Drop(seat, id) + if err != nil { + return nil, err + } + fmt.Fprintf(os.Stderr, "transfer %s of %s dropped by hand by %s: %s\n", id, seat, stdio.CallerOf(args), str(args, "why")) + return map[string]any{"dropped": t.ID}, nil + } + return h.IncomingOf(seat) + }, + } +} + +// incomingCap reads incoming's cap from its binding (the bus's bound of the object store), once it answers one. +func incomingCap() func() (uint64, error) { + var known atomic.Uint64 + return func() (uint64, error) { + if c := known.Load(); c > 0 { + return c, nil + } + f, err := busstate.ObjectsOf(packageseats.IncomingObjects).Fill() + if err != nil { + return 0, fmt.Errorf("incoming's cap cannot be read from its binding: %w", err) + } + if f.MaxBytes == 0 { + return 0, errors.New("incoming's binding carries no cap") + } + known.Store(f.MaxBytes) + return f.MaxBytes, nil + } +} + +// armIdle writes a transfer's idle key with its own time to live; one already there is the same bound, kept. +func armIdle(key string, ttl time.Duration) error { + if _, err := stdio.State(packageseats.IdleState).CreateTTL(key, map[string]any{"transfer": key}, ttl); err != nil && !errors.Is(err, stdio.ErrChanged) { + return err + } + return nil +} + +// followIdle drops a transfer when its idle key ends (packageseats.Holder.Expired): the bus's marker is the event. +func followIdle(holder func() (*packageseats.Holder, error)) { + <-stdio.Served() + stdio.State(packageseats.IdleState).Follow("", func(ch stdio.StateChange) error { + if ch.Op != stdio.OpExpire { + return nil + } + h, err := holder() + if err != nil { + fmt.Fprintf(os.Stderr, "transfer %s's idle bound ended, but the holder cannot be made: %v\n", ch.Key, err) + return err + } + dropped, err := h.Expired(ch.Key) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return err + } + if dropped { + fmt.Fprintf(os.Stderr, "transfer %s had no chunk for %s: dropped with its chunks\n", ch.Key, packageseats.TransferIdle) + } + return nil + }) +} diff --git a/modules/gitea/cmd/package-registry/main.go b/modules/gitea/cmd/package-registry/main.go index 9f1ca0fc..0869cfca 100644 --- a/modules/gitea/cmd/package-registry/main.go +++ b/modules/gitea/cmd/package-registry/main.go @@ -21,11 +21,13 @@ import ( // ToolNames are the tools this bundle serves. The module's manifest lists no `tools`: its TypeScript // bundle's tools are served without one, and a claim without `serves` would offer a `tools` list to both // of the module's seats as their verbs. The README names these two, and a test holds the two together. -var ToolNames = []string{"go_packages", "go_retention", "npm_packages", "npm_retention"} +var ToolNames = []string{"go_packages", "go_retention", "go_incoming", "npm_packages", "npm_retention", "npm_incoming"} func main() { forge := func() (*Forge, error) { return ForgeFromEnv() } - tools := append(Tools(forge), SeatTools(func() (*packageseats.Holder, error) { return busHolder(forge) })...) + holder := func() (*packageseats.Holder, error) { return busHolder(forge) } + go followIdle(holder) + tools := append(Tools(forge, holder), SeatTools(holder)...) if err := stdio.Serve("", tools); err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) @@ -34,8 +36,9 @@ func main() { // Tools are the bundle's tools over the forge the function gives; asked per call, so a manifest that // does not set the environment is a refusal naming what is missing, not a bundle that will not start. -func Tools(forge func() (*Forge, error)) []stdio.Tool { - return append(typeTools(TypeGo, forge), typeTools(TypeNpm, forge)...) +func Tools(forge func() (*Forge, error), holder func() (*packageseats.Holder, error)) []stdio.Tool { + out := append(typeTools(TypeGo, forge), incomingTool(TypeGo, packageseats.GoSeat, holder)) + return append(append(out, typeTools(TypeNpm, forge)...), incomingTool(TypeNpm, packageseats.NpmSeat, holder)) } // typeTools are one package type's two tools, _packages and _retention, over the forge reading that type. diff --git a/modules/gitea/cmd/package-registry/retention_test.go b/modules/gitea/cmd/package-registry/retention_test.go index 7bbdbbd1..092d3dc1 100644 --- a/modules/gitea/cmd/package-registry/retention_test.go +++ b/modules/gitea/cmd/package-registry/retention_test.go @@ -402,12 +402,12 @@ func TestAWrongPasswordIsAFailureThatDoesNotSayThePassword(t *testing.T) { } func TestToolsAreTheOnesTheReadmeNamesAndSayWhatTheyDo(t *testing.T) { - tools := Tools(func() (*Forge, error) { return nil, fmt.Errorf("unset") }) + tools := Tools(func() (*Forge, error) { return nil, fmt.Errorf("unset") }, func() (*packageseats.Holder, error) { return nil, fmt.Errorf("unset") }) var served []string for _, tl := range tools { served = append(served, tl.Name) suffix := "(r)" - if strings.HasSuffix(tl.Name, "_retention") { + if strings.HasSuffix(tl.Name, "_retention") || strings.HasSuffix(tl.Name, "_incoming") { suffix = "(a)" } if !strings.HasSuffix(tl.Description, suffix) { @@ -468,11 +468,15 @@ func TestToolsAreTheOnesTheReadmeNamesAndSayWhatTheyDo(t *testing.T) { t.Errorf("the seat tools are %v", names) } req, _ := json.Marshal(m.Requires) - for _, want := range []string{`"name":"incoming"`, `"name":"transfers"`, `"name":"findings"`} { + for _, want := range []string{`"name":"incoming"`, `"name":"transfers"`, `"name":"findings"`, `"name":"idle","options":{"key-ttl":true}`} { if !strings.Contains(string(req), want) { - t.Errorf("the manifest does not require %s", want) + t.Errorf("the manifest does not require %s (%s)", want, req) } } + // transfers keeps for good: a lifetime of its own would drop a transfer's key and leave its chunks in incoming. + if strings.Contains(string(req), `"name":"transfers","options"`) { + t.Errorf("transfers has options (a lifetime): %s", req) + } // The retention asks the builder's versions; the forge's other bundles ask the indexer's modules and record a re-read. if strings.Join(m.Invokes, ",") != "mesh-delivery-indexer.indexer_read_modules,seat:mesh-hand-acts.hand-act,mesh-delivery-builder.builder_versions" { t.Errorf("the manifest invokes %v", m.Invokes) diff --git a/modules/gitea/cmd/package-registry/seats.go b/modules/gitea/cmd/package-registry/seats.go index 759bdfad..6447791f 100644 --- a/modules/gitea/cmd/package-registry/seats.go +++ b/modules/gitea/cmd/package-registry/seats.go @@ -263,6 +263,9 @@ func finding(kind, subject, words string, raised bool) { } } +// boundCap is incoming's cap from its binding, read once per bundle. +var boundCap = incomingCap() + // busHolder is the holder over the runtime's bus state, the builder's numbering and the forge. func busHolder(forge func() (*Forge, error)) (*packageseats.Holder, error) { callers, err := packageseats.ConfiguredCallers() @@ -271,7 +274,8 @@ func busHolder(forge func() (*Forge, error)) (*packageseats.Holder, error) { } return &packageseats.Holder{Callers: callers, Objects: parts.BusObjects(packageseats.IncomingObjects), Transfers: parts.BusTransfers[packageseats.Opening](packageseats.TransfersState), - Builder: delivery.BuilderVersions{Ask: stdio.Ask}, Store: seatStore{forge}, Say: say, Finding: finding}, nil + Builder: delivery.BuilderVersions{Ask: stdio.Ask}, Store: seatStore{forge}, Say: say, Finding: finding, + Cap: boundCap, Idle: armIdle}, nil } // SeatTools are both seats' verbs as tools .. diff --git a/modules/gitea/cmd/package-registry/whole_integration_test.go b/modules/gitea/cmd/package-registry/whole_integration_test.go index 9085bd12..401adb78 100644 --- a/modules/gitea/cmd/package-registry/whole_integration_test.go +++ b/modules/gitea/cmd/package-registry/whole_integration_test.go @@ -114,6 +114,7 @@ func (p *player) handle(m wire) { Subject string `json:"subject"` Body json.RawMessage `json:"body"` Kind string `json:"kind"` + TTL int `json:"ttl_seconds"` } _ = json.Unmarshal(m.Params, &a) switch { @@ -131,7 +132,7 @@ func (p *player) handle(m wire) { case strings.HasPrefix(m.Method, "mesh/objects."): p.objects(m, a.Objects, a.Name, a.Data) case strings.HasPrefix(m.Method, "mesh/state."): - p.state(m, a.State, a.Key, a.Value) + p.state(m, a.State, a.Key, a.Value, a.TTL) default: p.answer(m.ID, nil, errors.New("not played here: "+m.Method)) } @@ -172,7 +173,7 @@ func (p *player) objects(m wire, local, name string, data []byte) { } } -func (p *player) state(m wire, state, key string, value json.RawMessage) { +func (p *player) state(m wire, state, key string, value json.RawMessage, ttl int) { kv, err := p.js.KeyValue(p.ctx, "gitea_"+state) if err != nil { p.answer(m.ID, nil, err) @@ -193,6 +194,13 @@ func (p *player) state(m wire, state, key string, value json.RawMessage) { case "mesh/state.put": rev, err := kv.Put(p.ctx, key, value) p.answer(m.ID, map[string]any{"revision": rev}, err) + case "mesh/state.create": + var opts []jetstream.KVCreateOpt + if ttl > 0 { + opts = append(opts, jetstream.KeyTTL(time.Duration(ttl)*time.Second)) + } + rev, err := kv.Create(p.ctx, key, value, opts...) + p.answer(m.ID, map[string]any{"revision": rev}, err) case "mesh/state.delete": p.answer(m.ID, map[string]any{}, kv.Delete(p.ctx, key)) case "mesh/state.keys": @@ -268,6 +276,10 @@ func TestThePackageSeatsWhole(t *testing.T) { t.Fatal(err) } } + // idle takes keys with their own time to live (key-ttl), as the bus module makes it. + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "gitea_" + packageseats.IdleState, LimitMarkerTTL: time.Minute}); err != nil { + t.Fatal(err) + } if _, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "gitea_" + packageseats.IncomingObjects}); err != nil { t.Fatal(err) } @@ -364,6 +376,11 @@ func TestThePackageSeatsWhole(t *testing.T) { if transfer == "" || len(lacks) != 1 || lacks[0] != digest { t.Fatalf("publish answered transfer %q lacks %v", transfer, lacks) } + // The open transfer is bounded by its idle key, written through the runtime with its own time to live. + idle, _ := js.KeyValue(ctx, "gitea_"+packageseats.IdleState) + if _, err := idle.Get(ctx, packageseats.GoSeat+"_"+transfer); err != nil { + t.Errorf("the open transfer %s has no idle key: %v", transfer, err) + } if err := send(transfer, digest, zipped); err != nil { t.Fatal(err) } diff --git a/modules/gitea/internal/packageseats/holder.go b/modules/gitea/internal/packageseats/holder.go index 7a3706a3..80ae121e 100644 --- a/modules/gitea/internal/packageseats/holder.go +++ b/modules/gitea/internal/packageseats/holder.go @@ -44,8 +44,8 @@ const ( ) // The bounds of incoming (design 51 *Publishing*, ADR 0611): a chunk at most 1 MiB, a transfer with no chunk for 24 -// hours dropped, a finding past 80 % of the cap. A package is one file, far smaller than an image, so the cap is the -// manifest's 2 GiB. +// hours dropped, a finding past 80 % of the cap. The cap is the one incoming's binding carries; IncomingCap is the +// manifest's 2 GiB (a package is one file, far smaller than an image), said only where no binding is read. const ( MaxChunk = 1 << 20 IncomingCap = 2 << 30 @@ -53,6 +53,9 @@ const ( IncomingFullPct = 80 IncomingObjects = "incoming" TransfersState = "transfers" + // IdleState is the state of one key per open transfer with its own time to live (a state declared key-ttl): its + // end is the event that drops a transfer left with no chunk for TransferIdle, never a timer. + IdleState = "idle" // KindPackage is the one kind a package seat holds. KindPackage = "package" // Self is the forge module, the holder every outcome names. @@ -151,7 +154,20 @@ type Holder struct { // Finding writes (raised) or clears the module's finding of a kind for a subject. Finding func(kind, subject, words string, raised bool) Now func() time.Time - Cap uint64 + // Cap reads incoming's cap from its binding; IncomingCap when nil or when the binding names none. + Cap func() (uint64, error) + // Idle writes a transfer's idle key with its own time to live, once; one already there is no fault. Nil arms nothing. + Idle func(key string, ttl time.Duration) error +} + +// CapNow is incoming's cap as its binding carries it. +func (h *Holder) CapNow() uint64 { + if h.Cap != nil { + if c, err := h.Cap(); err == nil && c > 0 { + return c + } + } + return IncomingCap } func (h *Holder) now() time.Time { @@ -168,10 +184,7 @@ func Reference(name, version string) string { return name + "@" + version } func key(seat, transfer string) string { return seat + "_" + transfer } func (h *Holder) receiver() *parts.Receiver[Opening] { - cap := h.Cap - if cap == 0 { - cap = IncomingCap - } + cap := h.CapNow() return &parts.Receiver[Opening]{Objects: h.Objects, Transfers: h.Transfers, Cap: cap, MaxChunk: MaxChunk, Idle: TransferIdle, Full: IncomingFullPct, Now: h.now, // The builder's numbering is asked before anything is uploaded (ADR 0605 ยง4). @@ -230,7 +243,6 @@ func (h *Holder) Publish(seat, caller string, p Publish) (Answer, error) { if err := checkPublish(seat, p); err != nil { return Answer{}, err } - _, _ = h.receiver().DropIdle() // A rebuild of a published version answers it held and uploads nothing; the same version with other content is // refused, as the registry would refuse it. if d, ok, err := h.Store.Held(seat, p.Name, p.Version); err != nil { @@ -253,6 +265,13 @@ func (h *Holder) Publish(seat, caller string, p Publish) (Answer, error) { if err != nil { return h.answer(p.Transfer, got), fmt.Errorf("publish: %w", err) } + // An open transfer is bounded by its idle key; one that cannot be written fails the publish loudly, and the build + // module's next publish of the same transfer answers the part held and arms it again. + if !got.Ended && h.Idle != nil { + if err := h.Idle(key(seat, p.Transfer), TransferIdle); err != nil { + return h.answer(p.Transfer, got), fmt.Errorf("publish: transfer %s is open but its idle bound is not kept, so it would never be dropped: %w", p.Transfer, err) + } + } return h.answer(p.Transfer, got), nil } @@ -324,11 +343,65 @@ func (h *Holder) Retained(seat, name, version, digest string) error { return h.Say(seat, EventRemoved, Outcome{Name: name, Version: version, Kind: KindPackage, Digest: digest, Reason: ReasonRetention}) } -// Drop drops a transfer by hand; DropIdle every transfer idle for TransferIdle. +// Drop drops a transfer by hand. func (h *Holder) Drop(seat, transfer string) (Transfer, error) { return h.receiver().Drop(key(seat, transfer)) } +// Incoming is what npm_incoming and go_incoming answer of a seat: its open transfers, incoming's fill (both seats +// share the store) and the cap its binding carries. +type Incoming struct { + Seat string `json:"seat"` + Transfers []Transfer `json:"transfers"` + Used uint64 `json:"used"` + Cap uint64 `json:"cap"` +} + +// IncomingOf reads a seat's open transfers and incoming's fill. +func (h *Holder) IncomingOf(seat string) (Incoming, error) { + if _, ok := Seats[seat]; !ok { + return Incoming{}, fmt.Errorf("the forge holds no seat %s", seat) + } + all, err := h.Transfers.List() + if err != nil { + return Incoming{}, fmt.Errorf("the transfers cannot be read: %w", err) + } + out := Incoming{Seat: seat, Transfers: []Transfer{}, Cap: h.CapNow()} + for _, t := range all { + if t.Shape.Seat == seat { + out.Transfers = append(out.Transfers, t) + } + } + if out.Used, err = h.Objects.Used(); err != nil { + return out, fmt.Errorf("incoming's fill cannot be read: %w", err) + } + return out, nil +} + +// Expired is told that a transfer's idle key (its key in the state transfers) ended: a transfer with no chunk for +// TransferIdle is dropped with its chunks; one that took a chunk since is armed again for the rest; one gone is +// nothing to do. It answers whether it dropped it. +func (h *Holder) Expired(id string) (bool, error) { + t, found, err := h.Transfers.Get(id) + if err != nil { + return false, fmt.Errorf("transfer %s's idle bound ended, but it cannot be read: %w", id, err) + } + if !found { + return false, nil + } + idle := h.now().Sub(t.Last) + if idle >= TransferIdle { + if _, err := h.receiver().Drop(id); err != nil { + return false, fmt.Errorf("transfer %s had no chunk for %s and is not dropped: %w", id, TransferIdle, err) + } + return true, nil + } + if h.Idle == nil { + return false, nil + } + return false, h.Idle(id, max(TransferIdle-idle, time.Second)) +} + func checkPublish(seat string, p Publish) error { switch { case Seats[seat] == "": diff --git a/modules/gitea/internal/packageseats/holder_test.go b/modules/gitea/internal/packageseats/holder_test.go index 41aeaae9..700001c0 100644 --- a/modules/gitea/internal/packageseats/holder_test.go +++ b/modules/gitea/internal/packageseats/holder_test.go @@ -149,6 +149,48 @@ func TestTheForgeHoldsBothPackageSeats(t *testing.T) { } } }) + // Tools review F4 and running review F2's kind: a seat's incoming is read per seat with the cap its binding + // carries; an open transfer is armed with its idle key, re-armed for the rest when it took a chunk since, and + // dropped with its chunks when it had none for 24 hours; a cap from the binding refuses a transfer past it. + t.Run("incoming: read per seat, the binding's cap, an idle transfer dropped when its key ends", func(t *testing.T) { + b := newBed(true) + now := time.Unix(1e9, 0) + b.h.Now = func() time.Time { return now } + armed := map[string]time.Duration{} + b.h.Idle = func(k string, ttl time.Duration) error { armed[k] = ttl; return nil } + b.h.Cap = func() (uint64, error) { return 1000, nil } + big := []byte(strings.Repeat("x", 600)) + if _, err := b.h.Publish(NpmSeat, nodeBuild, pub("a", "0.1.0", big)); err != nil { + t.Fatal(err) + } + if _, err := b.h.Publish(GoSeat, goBuild, pub("b", "0.1.0", big)); err != nil || armed["go-package-registry_t-b"] != TransferIdle { + t.Fatalf("publish = %v, armed %v; want the Go transfer's idle key for %s", err, armed, TransferIdle) + } + _, _ = b.h.PublishPart(NpmSeat, nodeBuild, "t-a", digest(big), 0, false, big[:500]) + in, err := b.h.IncomingOf(NpmSeat) + if err != nil || len(in.Transfers) != 1 || in.Transfers[0].Shape.Name != "a" || in.Cap != 1000 || in.Used != 500 { + t.Fatalf("IncomingOf(npm) = %+v, %v; want a's transfer alone, 500 used of the binding's 1000", in, err) + } + if _, err := b.h.Publish(NpmSeat, nodeBuild, pub("c", "0.1.0", big)); err == nil || !strings.Contains(err.Error(), "cap") { + t.Errorf("a transfer past the binding's cap = %v; want refused at the cap", err) + } + now = now.Add(time.Hour) + _, _ = b.h.PublishPart(GoSeat, goBuild, "t-b", digest(big), 0, false, big[:10]) + now = now.Add(TransferIdle - time.Hour) + if dropped, err := b.h.Expired("go-package-registry_t-b"); err != nil || dropped || armed["go-package-registry_t-b"] != time.Hour { + t.Errorf("b's key ends an hour after a chunk: %v, %v, armed %v; want kept, armed 1h", dropped, err, armed) + } + if dropped, err := b.h.Expired("npm-package-registry_t-a"); err != nil || !dropped { + t.Errorf("a's key ends 24 hours after its last chunk: %v, %v; want dropped", dropped, err) + } + if in, _ := b.h.IncomingOf(NpmSeat); len(in.Transfers) != 0 || in.Used != 10 { + t.Errorf("after a's drop: %+v; want no npm transfer and only b's 10 bytes used", in) + } + b.h.Idle = func(string, time.Duration) error { return errors.New("bus down") } + if _, err := b.h.Publish(NpmSeat, nodeBuild, pub("d", "0.1.0", file)); err == nil || !strings.Contains(err.Error(), "idle bound") { + t.Errorf("a publish whose idle key is not written = %v; want it failed loudly", err) + } + }) t.Run("refusals", func(t *testing.T) { for _, c := range []struct { name string diff --git a/modules/gitea/module.json b/modules/gitea/module.json index 0caba4c2..0c0e1095 100644 --- a/modules/gitea/module.json +++ b/modules/gitea/module.json @@ -17,9 +17,14 @@ { "name": "transfers", "storage": "key-value", + "template": "state" + }, + { + "name": "idle", + "storage": "key-value", "template": "state", "options": { - "ttl-seconds": 86400 + "key-ttl": true } }, {