diff --git a/CONFORMANCE.md b/CONFORMANCE.md index b22a8e0..72c7093 100644 --- a/CONFORMANCE.md +++ b/CONFORMANCE.md @@ -37,6 +37,7 @@ Statuses are stated per the architecture's honesty rule. Built member-layer work - **Covenant and economy are stubs on main.** Their package docs carry the binding invariants; the built implementations are in the open, stacked PRs and must satisfy those docs to merge. - **Record domain-tag rename** (product naming) is pending and must land **before any durable persistence** — retagging after durable logs exist would be a rewrite. - **Named record residuals** (steganographic floor, timestamp covert channel, witness amnesia, liveness gap) are documented in the record package doc rather than pretended away. +- **Contributor node — Executor seam built; runtime implementations pending.** `internal/contribute` fixes the node-contribution contract (Executor / ServiceExecutor / Registry / NodeContract) with two invariants enforced by tests: a scavenged node accepts only preemptible executors (owner-never-interrupted), and a Service workload requires pinned placement. `cmd/cloudy-agent` runs it end to end with a **real** Storage executor (opaque sealed shards + the `internal/storage` proof-of-possession) and **placeholder** Compute/Service executors. The heavy runtime port from the coordinator's transitional `internal/agent` — Docker executor, gopsutil hardware sampling, opt-out/allowlist/printers, and the agent→cloudyd→`/v0` relay — is the named next step and lands *behind* this seam without changing it. ## Dependency declaration diff --git a/cmd/cloudy-agent/agent.go b/cmd/cloudy-agent/agent.go new file mode 100644 index 0000000..8b2aeb1 --- /dev/null +++ b/cmd/cloudy-agent/agent.go @@ -0,0 +1,78 @@ +package main + +import ( + "context" + "errors" + "log/slog" + + "github.com/NTARI-RAND/Cloudy/internal/contribute" +) + +// ErrOwnerActive is returned when scavenged work is refused because the owner +// has returned. It is a deferral, not a failure. +var ErrOwnerActive = errors.New("cloudy-agent: owner active; scavenged work deferred") + +// Agent is the contributor daemon's control loop over the contribute seam. It +// owns the owner-never-interrupted gate: on a scavenged node it stops +// advertising and dispatching work while the owner is active. On a pinned node +// the owner signal does not gate anything — the capacity is dedicated. +// +// This is the seam-side skeleton the migrated SoHoLINK agent plugs into: the +// heartbeat/opt-out/telemetry loop calls Capabilities and Dispatch, and the +// Docker/gopsutil/spiffe code lands as Executor implementations and a real +// ownerActive probe, not as changes here. +type Agent struct { + reg *contribute.Registry + ownerActive func() bool + log *slog.Logger +} + +// NewAgent builds an agent. ownerActive reports whether the machine's owner is +// currently using it (nil means "never active", e.g. a headless pinned box). +func NewAgent(reg *contribute.Registry, ownerActive func() bool, log *slog.Logger) *Agent { + if ownerActive == nil { + ownerActive = func() bool { return false } + } + if log == nil { + log = slog.Default() + } + return &Agent{reg: reg, ownerActive: ownerActive, log: log} +} + +// scavengedYield reports whether the owner gate is currently closing this node +// to scavenged work. +func (a *Agent) scavengedYield() bool { + return a.reg.Contract().Placement == contribute.Scavenged && a.ownerActive() +} + +// Capabilities is what the agent advertises upstream this heartbeat. A +// scavenged node advertises nothing while the owner is active — it has yielded +// the machine entirely. +func (a *Agent) Capabilities() []contribute.WorkloadKind { + if a.scavengedYield() { + return nil + } + return a.reg.Capabilities() +} + +// Dispatch runs a one-shot job, refusing scavenged work while the owner is +// active (ErrOwnerActive). +func (a *Agent) Dispatch(ctx context.Context, job contribute.AssignedJob) (contribute.Result, error) { + if a.scavengedYield() { + return contribute.Result{JobID: job.ID, OK: false, Detail: "owner active"}, ErrOwnerActive + } + return a.reg.Dispatch(ctx, job) +} + +// ServeService runs the node's pinned service (if any) until ctx is cancelled. +// Services run on pinned capacity, so the owner gate does not apply. +func (a *Agent) ServeService(ctx context.Context, job contribute.AssignedJob) error { + svc, ok := a.reg.Service() + if !ok { + return contribute.ErrNoExecutor + } + a.log.Info("service starting", "node", a.reg.Contract().NodeID, "job", job.ID) + err := svc.Serve(ctx, job) + a.log.Info("service stopped", "node", a.reg.Contract().NodeID, "job", job.ID, "err", err) + return err +} diff --git a/cmd/cloudy-agent/agent_test.go b/cmd/cloudy-agent/agent_test.go new file mode 100644 index 0000000..0b6cb2d --- /dev/null +++ b/cmd/cloudy-agent/agent_test.go @@ -0,0 +1,128 @@ +package main + +import ( + "context" + "encoding/base64" + "encoding/json" + "errors" + "testing" + + "github.com/NTARI-RAND/Cloudy/internal/contribute" +) + +func pinnedAgent(t *testing.T, ownerActive func() bool) *Agent { + t.Helper() + reg, err := contribute.NewRegistry(contribute.NodeContract{ + NodeID: "test", Placement: contribute.Pinned, + Kinds: []contribute.WorkloadKind{contribute.Compute, contribute.Storage, contribute.Service}, + }) + if err != nil { + t.Fatal(err) + } + if err := reg.Register(echoExecutor{}); err != nil { + t.Fatal(err) + } + if err := reg.Register(newStorageExecutor()); err != nil { + t.Fatal(err) + } + if err := reg.RegisterService(&presenceService{}); err != nil { + t.Fatal(err) + } + return NewAgent(reg, ownerActive, nil) +} + +func TestStorageStoreThenAuditProvesPossession(t *testing.T) { + agent := pinnedAgent(t, nil) + shard := []byte("opaque sealed bytes the host can never read but can prove it holds") + + store, _ := json.Marshal(storageOp{Op: "store", ID: "o1", Data: base64.StdEncoding.EncodeToString(shard)}) + res, err := agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "s1", Kind: contribute.Storage, Payload: store}) + if err != nil || !res.OK { + t.Fatalf("store: res=%+v err=%v", res, err) + } + + audit, _ := json.Marshal(storageOp{Op: "audit", ID: "o1"}) + res, err = agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "s2", Kind: contribute.Storage, Payload: audit}) + if err != nil || !res.OK { + t.Fatalf("audit should prove possession: res=%+v err=%v", res, err) + } + + // Auditing an unknown shard fails. + missing, _ := json.Marshal(storageOp{Op: "audit", ID: "nope"}) + if _, err := agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "s3", Kind: contribute.Storage, Payload: missing}); err == nil { + t.Fatal("audit of missing shard should error") + } +} + +func TestComputeEchoRoutes(t *testing.T) { + agent := pinnedAgent(t, nil) + res, err := agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "c1", Kind: contribute.Compute, Payload: []byte("ping")}) + if err != nil || !res.OK || string(res.Output) != "ping" { + t.Fatalf("compute echo: res=%+v err=%v", res, err) + } +} + +func TestPinnedNodeIgnoresOwnerGate(t *testing.T) { + agent := pinnedAgent(t, func() bool { return true }) // owner "active" + // Pinned capacity is dedicated: owner activity does not reduce it. + if caps := agent.Capabilities(); len(caps) != 3 { + t.Fatalf("pinned node should advertise all 3 kinds regardless of owner, got %v", caps) + } + if _, err := agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "c1", Kind: contribute.Compute, Payload: []byte("x")}); err != nil { + t.Fatalf("pinned dispatch should not be gated: %v", err) + } +} + +func TestScavengedNodeYieldsToOwner(t *testing.T) { + reg, err := contribute.NewRegistry(contribute.NodeContract{ + NodeID: "laptop", Placement: contribute.Scavenged, + Kinds: []contribute.WorkloadKind{contribute.Compute, contribute.Storage}, + }) + if err != nil { + t.Fatal(err) + } + if err := reg.Register(echoExecutor{}); err != nil { + t.Fatal(err) + } + if err := reg.Register(newStorageExecutor()); err != nil { + t.Fatal(err) + } + + active := true + agent := NewAgent(reg, func() bool { return active }, nil) + + // Owner present: advertise nothing, refuse work. + if caps := agent.Capabilities(); caps != nil { + t.Fatalf("scavenged node with owner active should advertise nothing, got %v", caps) + } + if _, err := agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "c1", Kind: contribute.Compute, Payload: []byte("x")}); !errors.Is(err, ErrOwnerActive) { + t.Fatalf("want ErrOwnerActive, got %v", err) + } + + // Owner leaves: capacity returns. + active = false + if caps := agent.Capabilities(); len(caps) != 2 { + t.Fatalf("scavenged node with owner away should advertise 2 kinds, got %v", caps) + } + if res, err := agent.Dispatch(context.Background(), contribute.AssignedJob{ID: "c2", Kind: contribute.Compute, Payload: []byte("go")}); err != nil || !res.OK { + t.Fatalf("dispatch after owner leaves: res=%+v err=%v", res, err) + } +} + +func TestServiceLifecycle(t *testing.T) { + agent := pinnedAgent(t, nil) + svc, ok := agent.reg.Service() + if !ok { + t.Fatal("expected a registered service") + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- agent.ServeService(ctx, contribute.AssignedJob{ID: "svc", Kind: contribute.Service}) }() + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("service should stop cleanly on cancel, got %v", err) + } + if svc.Healthy() { + t.Error("service should report unhealthy after shutdown") + } +} diff --git a/cmd/cloudy-agent/executors.go b/cmd/cloudy-agent/executors.go new file mode 100644 index 0000000..b659349 --- /dev/null +++ b/cmd/cloudy-agent/executors.go @@ -0,0 +1,49 @@ +package main + +import ( + "context" + "sync" + + "github.com/NTARI-RAND/Cloudy/internal/contribute" +) + +// echoExecutor is a minimal, dependency-free Compute executor: it returns its +// payload. It stands in for the Docker compute executor (the heavy port that +// satisfies this same seam next) so the Compute route is proven without +// dragging the docker toolchain into the module yet. Preemptible. +type echoExecutor struct{} + +func (echoExecutor) Kind() contribute.WorkloadKind { return contribute.Compute } +func (echoExecutor) Preemptible() bool { return true } +func (echoExecutor) Run(_ context.Context, job contribute.AssignedJob) (contribute.Result, error) { + return contribute.Result{JobID: job.ID, OK: true, Detail: "echo", Output: job.Payload}, nil +} + +// presenceService is a demo long-running Service: it reports healthy for its +// lifetime and shuts down cleanly when the context is cancelled. It exists to +// prove the ServiceExecutor half of the seam and the pinned-capacity contract; +// a real hosted service (a site, a store, a game server) implements the same +// interface. +type presenceService struct { + mu sync.Mutex + healthy bool +} + +func (p *presenceService) Kind() contribute.WorkloadKind { return contribute.Service } + +func (p *presenceService) Healthy() bool { + p.mu.Lock() + defer p.mu.Unlock() + return p.healthy +} + +func (p *presenceService) Serve(ctx context.Context, _ contribute.AssignedJob) error { + p.mu.Lock() + p.healthy = true + p.mu.Unlock() + <-ctx.Done() + p.mu.Lock() + p.healthy = false + p.mu.Unlock() + return ctx.Err() +} diff --git a/cmd/cloudy-agent/main.go b/cmd/cloudy-agent/main.go new file mode 100644 index 0000000..ce08392 --- /dev/null +++ b/cmd/cloudy-agent/main.go @@ -0,0 +1,130 @@ +// Command cloudy-agent is the contributor daemon a Cloudy member runs on a +// machine that contributes capacity. It is the home the SoHoLINK internal/agent +// code moves into (executor, allowlist, opt-out, printers, hardware sampling, +// heartbeat, telemetry) — but built around the contribute seam so the runtime +// implementations (Docker, gopsutil, an Android storage provider, GPU) are +// pluggable Executors, not the definition. +// +// This entrypoint is honest about what it is today: the seam and control loop +// are real and tested; the registered executors are a real Storage node (over +// internal/storage's proof-of-possession), a placeholder Compute echo standing +// in for the Docker executor, and a demo presence Service. The Docker/gopsutil/ +// spiffe port and the agent->cloudyd->/v0 relay are the named next steps; they +// land behind this seam without changing it. +package main + +import ( + "context" + "encoding/base64" + "encoding/json" + "flag" + "log/slog" + "os" + "os/signal" + "time" + + "github.com/NTARI-RAND/Cloudy/internal/contribute" +) + +func main() { + var ( + nodeID = flag.String("node", "cloudy-node-0", "node identifier") + pinned = flag.Bool("pinned", true, "pinned (dedicated) capacity; false = scavenged (idle-first)") + withSvc = flag.Bool("service", true, "offer the demo presence service (requires -pinned)") + selftest = flag.Bool("selftest", true, "run a store->audit->echo self-test and exit") + ) + flag.Parse() + + log := slog.New(slog.NewTextHandler(os.Stderr, nil)) + + placement := contribute.Scavenged + kinds := []contribute.WorkloadKind{contribute.Compute, contribute.Storage} + if *pinned { + placement = contribute.Pinned + if *withSvc { + kinds = append(kinds, contribute.Service) + } + } + + reg, err := contribute.NewRegistry(contribute.NodeContract{NodeID: *nodeID, Placement: placement, Kinds: kinds}) + if err != nil { + log.Error("contract invalid", "err", err) + os.Exit(1) + } + if err := reg.Register(echoExecutor{}); err != nil { + log.Error("register compute", "err", err) + os.Exit(1) + } + if err := reg.Register(newStorageExecutor()); err != nil { + log.Error("register storage", "err", err) + os.Exit(1) + } + if *pinned && *withSvc { + if err := reg.RegisterService(&presenceService{}); err != nil { + log.Error("register service", "err", err) + os.Exit(1) + } + } + + // ownerActive: no real hardware probe yet (that is the gopsutil port); + // pinned nodes are never gated, so this stub is correct for -pinned. + agent := NewAgent(reg, func() bool { return false }, log) + log.Info("cloudy-agent up", "node", *nodeID, "placement", placement.String(), "capabilities", capsToStrings(agent.Capabilities())) + + if *selftest { + runSelfTest(context.Background(), agent, log) + return + } + + // Long-running mode: serve the pinned service until interrupted. + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) + defer stop() + if _, ok := reg.Service(); ok { + if err := agent.ServeService(ctx, contribute.AssignedJob{ID: "presence-0", Kind: contribute.Service}); err != nil && ctx.Err() == nil { + log.Error("service failed", "err", err) + os.Exit(1) + } + } else { + <-ctx.Done() + } +} + +func capsToStrings(ks []contribute.WorkloadKind) []string { + out := make([]string, len(ks)) + for i, k := range ks { + out[i] = k.String() + } + return out +} + +// runSelfTest exercises the seam end to end: store an opaque shard, prove +// possession, and echo a compute job. +func runSelfTest(ctx context.Context, agent *Agent, log *slog.Logger) { + shard := []byte("opaque-sealed-shard-bytes-the-node-cannot-read") + storePayload, _ := json.Marshal(storageOp{Op: "store", ID: "obj-1", Data: base64.StdEncoding.EncodeToString(shard)}) + if res, err := agent.Dispatch(ctx, contribute.AssignedJob{ID: "s1", Kind: contribute.Storage, Payload: storePayload}); err != nil { + log.Error("store failed", "err", err) + } else { + log.Info("store", "ok", res.OK, "detail", res.Detail) + } + + auditPayload, _ := json.Marshal(storageOp{Op: "audit", ID: "obj-1"}) + if res, err := agent.Dispatch(ctx, contribute.AssignedJob{ID: "s2", Kind: contribute.Storage, Payload: auditPayload}); err != nil { + log.Error("audit failed", "err", err) + } else { + log.Info("audit", "ok", res.OK, "detail", res.Detail) + } + + if res, err := agent.Dispatch(ctx, contribute.AssignedJob{ID: "c1", Kind: contribute.Compute, Payload: []byte("hello")}); err != nil { + log.Error("compute failed", "err", err) + } else { + log.Info("compute", "ok", res.OK, "output", string(res.Output)) + } + + if svc, ok := agent.reg.Service(); ok { + sctx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) + defer cancel() + _ = agent.ServeService(sctx, contribute.AssignedJob{ID: "svc-1", Kind: contribute.Service}) + log.Info("service healthy after stop", "healthy", svc.Healthy()) + } +} diff --git a/cmd/cloudy-agent/storageexec.go b/cmd/cloudy-agent/storageexec.go new file mode 100644 index 0000000..eaf9f7e --- /dev/null +++ b/cmd/cloudy-agent/storageexec.go @@ -0,0 +1,95 @@ +package main + +import ( + "context" + "crypto/rand" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "sync" + + "github.com/NTARI-RAND/Cloudy/internal/contribute" + "github.com/NTARI-RAND/Cloudy/internal/storage" +) + +// storageExecutor is a real Storage-kind executor behind the contribute seam. +// It holds OPAQUE sealed shards — it can never read them, because the member +// seals client-side before the bytes ever leave the device (the zero-knowledge +// storage-privacy design). On an "audit" it proves it still holds the exact +// bytes using the proof-of-possession challenge/response in internal/storage, +// without learning anything about the content. +// +// Preemptible: yes. Holding bytes survives a pause, and the agent stops +// serving reads while the owner is active. (Durability across a real +// preemption needs on-disk persistence — the in-memory map here is the seam +// proof; persistence is the same follow-up the Drops log has.) +type storageExecutor struct { + mu sync.Mutex + shards map[string][]byte + audits int +} + +func newStorageExecutor() *storageExecutor { + return &storageExecutor{shards: make(map[string][]byte), audits: 8} +} + +func (s *storageExecutor) Kind() contribute.WorkloadKind { return contribute.Storage } +func (s *storageExecutor) Preemptible() bool { return true } + +// storageOp is the opaque job payload the seam hands this executor. +type storageOp struct { + Op string `json:"op"` // "store" | "audit" + ID string `json:"id"` // shard identifier + Data string `json:"data_b64,omitempty"` // opaque sealed bytes (store only) +} + +func (s *storageExecutor) Run(_ context.Context, job contribute.AssignedJob) (contribute.Result, error) { + var op storageOp + if err := json.Unmarshal(job.Payload, &op); err != nil { + return contribute.Result{JobID: job.ID}, fmt.Errorf("storage: bad payload: %w", err) + } + switch op.Op { + case "store": + blob, err := base64.StdEncoding.DecodeString(op.Data) + if err != nil { + return contribute.Result{JobID: job.ID}, fmt.Errorf("storage: bad data: %w", err) + } + if len(blob) == 0 { + return contribute.Result{JobID: job.ID}, errors.New("storage: empty shard") + } + s.mu.Lock() + s.shards[op.ID] = blob + n := len(blob) + s.mu.Unlock() + return contribute.Result{JobID: job.ID, OK: true, Detail: fmt.Sprintf("stored %d opaque bytes for %s", n, op.ID)}, nil + case "audit": + s.mu.Lock() + blob, ok := s.shards[op.ID] + s.mu.Unlock() + if !ok { + return contribute.Result{JobID: job.ID}, fmt.Errorf("storage: no shard %s", op.ID) + } + // In production the challenge table is precomputed on the member's + // device and only expected digests travel; here we build and check it + // in one place to exercise the primitive end to end. + table, err := storage.BuildChallengeTable(blob, s.audits, rand.Reader) + if err != nil { + return contribute.Result{JobID: job.ID}, err + } + ch, expected, err := table.Next() + if err != nil { + return contribute.Result{JobID: job.ID}, err + } + resp, err := storage.Respond(blob, ch) + if err != nil { + return contribute.Result{JobID: job.ID}, err + } + if !storage.VerifyProof(expected, resp) { + return contribute.Result{JobID: job.ID, OK: false, Detail: "possession proof did not verify"}, nil + } + return contribute.Result{JobID: job.ID, OK: true, Detail: fmt.Sprintf("possession proven for %s", op.ID)}, nil + default: + return contribute.Result{JobID: job.ID}, fmt.Errorf("storage: unknown op %q", op.Op) + } +} diff --git a/internal/contribute/contract.go b/internal/contribute/contract.go new file mode 100644 index 0000000..cc0764c --- /dev/null +++ b/internal/contribute/contract.go @@ -0,0 +1,59 @@ +package contribute + +import ( + "errors" + "fmt" +) + +var ( + // ErrNoKinds means a contract advertises no workload kinds. + ErrNoKinds = errors.New("contribute: node contract advertises no workload kinds") + // ErrServiceNeedsPinned means a Service kind was contracted on scavenged + // capacity — forbidden, because a service cannot yield to the owner. + ErrServiceNeedsPinned = errors.New("contribute: service workload requires pinned placement") + // ErrDupKind means a workload kind appears twice in a contract. + ErrDupKind = errors.New("contribute: duplicate workload kind in contract") +) + +// NodeContract is the owner's standing declaration: which workload kinds this +// machine accepts and on what terms (scavenged vs pinned). It is the local +// source of truth the Registry and agent enforce against; the wire never +// widens it. +type NodeContract struct { + NodeID string + Placement Placement + Kinds []WorkloadKind +} + +// Validate enforces the contract's internal invariants: at least one kind, no +// duplicates, all valid, and — the load-bearing rule — no Service on scavenged +// capacity. +func (c NodeContract) Validate() error { + if len(c.Kinds) == 0 { + return ErrNoKinds + } + seen := make(map[WorkloadKind]bool, len(c.Kinds)) + for _, k := range c.Kinds { + if !k.Valid() { + return fmt.Errorf("contribute: invalid workload kind %d", int(k)) + } + if seen[k] { + return fmt.Errorf("%w: %s", ErrDupKind, k) + } + seen[k] = true + } + if seen[Service] && c.Placement != Pinned { + return ErrServiceNeedsPinned + } + return nil +} + +// Allows reports whether k is within this contract. +func (c NodeContract) Allows(k WorkloadKind) bool { + for _, x := range c.Kinds { + if x == k { + return true + } + } + return false +} diff --git a/internal/contribute/contract_test.go b/internal/contribute/contract_test.go new file mode 100644 index 0000000..50215c0 --- /dev/null +++ b/internal/contribute/contract_test.go @@ -0,0 +1,69 @@ +package contribute + +import ( + "errors" + "testing" +) + +func TestWorkloadKindString(t *testing.T) { + cases := map[WorkloadKind]string{ + Compute: "compute", Print: "print", Storage: "storage", + Service: "service", WorkloadKind(99): "unknown", + } + for k, want := range cases { + if got := k.String(); got != want { + t.Errorf("WorkloadKind(%d).String() = %q, want %q", int(k), got, want) + } + } + if WorkloadKind(99).Valid() { + t.Error("WorkloadKind(99) should be invalid") + } + for _, k := range []WorkloadKind{Compute, Print, Storage, Service} { + if !k.Valid() { + t.Errorf("%s should be valid", k) + } + } +} + +func TestNodeContractValidate(t *testing.T) { + cases := []struct { + name string + c NodeContract + want error + }{ + {"empty", NodeContract{Placement: Pinned}, ErrNoKinds}, + {"invalid kind", NodeContract{Placement: Pinned, Kinds: []WorkloadKind{WorkloadKind(42)}}, nil}, // sentinel checked below + {"duplicate", NodeContract{Placement: Pinned, Kinds: []WorkloadKind{Compute, Compute}}, ErrDupKind}, + {"service scavenged", NodeContract{Placement: Scavenged, Kinds: []WorkloadKind{Service}}, ErrServiceNeedsPinned}, + {"service pinned ok", NodeContract{Placement: Pinned, Kinds: []WorkloadKind{Service}}, nil}, + {"scavenged compute ok", NodeContract{Placement: Scavenged, Kinds: []WorkloadKind{Compute, Storage}}, nil}, + {"pinned mixed ok", NodeContract{Placement: Pinned, Kinds: []WorkloadKind{Compute, Storage, Service}}, nil}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := tc.c.Validate() + if tc.name == "invalid kind" { + if err == nil { + t.Fatal("expected an error for an invalid kind") + } + return + } + if tc.want == nil && err != nil { + t.Fatalf("Validate() = %v, want nil", err) + } + if tc.want != nil && !errors.Is(err, tc.want) { + t.Fatalf("Validate() = %v, want %v", err, tc.want) + } + }) + } +} + +func TestNodeContractAllows(t *testing.T) { + c := NodeContract{Placement: Pinned, Kinds: []WorkloadKind{Compute, Storage}} + if !c.Allows(Compute) || !c.Allows(Storage) { + t.Error("contract should allow its own kinds") + } + if c.Allows(Service) || c.Allows(Print) { + t.Error("contract should not allow uncontracted kinds") + } +} diff --git a/internal/contribute/doc.go b/internal/contribute/doc.go new file mode 100644 index 0000000..8fa41d5 --- /dev/null +++ b/internal/contribute/doc.go @@ -0,0 +1,38 @@ +// Package contribute is the contributor-node Executor seam: the interface a +// Cloudy member machine implements to contribute capacity, decoupled from any +// one runtime. Docker containers, an Android storage provider, a long-running +// service, and GPU compute are all Executors behind this seam; none of them is +// the definition. The seam exists so the runtime-heavy implementations +// (docker/gopsutil/spiffe) plug into a contract that is already fixed and +// tested, rather than the contract being whatever today's Docker code happens +// to do. +// +// This package is NOT a JFA member-economy leaf. It never imports +// economy/covenant/record/dispute, and nothing in the import graph makes it a +// composition root. It is the node/contribution plane, which is a different +// concern from the member's credit/covenant/record world. +// +// Invariants this seam defends: +// +// - Owner is never interrupted. A node offered on SCAVENGED (idle-first) +// capacity may only host PREEMPTIBLE executors — work that can be paused or +// stopped the moment the owner returns. The Registry refuses to register a +// non-preemptible executor on a scavenged contract, and the agent control +// loop stops advertising and dispatching scavenged work while the owner is +// active. +// +// - A service is pinned, not scavenged. A long-running service cannot yield +// to the owner at 2pm, so it requires declared, dedicated (PINNED) +// capacity. NodeContract.Validate rejects a Service workload on a scavenged +// node; that is the structural line between "lend my idle desktop" and +// "dedicate this box to running a thing". +// +// - Opt-out is enforced locally, never trusted from the wire. Capabilities() +// advertises only what the node's own contract permits; the advertisement +// is a hint to the matcher, and the executor still refuses out-of-contract +// work locally. The coordinator is not a security boundary for opt-out. +// +// The seam carries no bytes of member data and speaks no protocol: it is the +// local dispatch contract. Wire relaying (agent signs node messages, cloudyd +// relays them over SCP /v0) sits above this package, per the Phase-1 design. +package contribute diff --git a/internal/contribute/executor.go b/internal/contribute/executor.go new file mode 100644 index 0000000..8d7388a --- /dev/null +++ b/internal/contribute/executor.go @@ -0,0 +1,115 @@ +package contribute + +import ( + "context" + "time" +) + +// WorkloadKind is what a contributor node can be asked to do. It mirrors the +// protocol's listing.WorkloadOptIn (Compute, Print, Storage) and adds Service +// — a long-running workload that is a first-class seam variant, not a batch +// job dressed up as one. The node contract decides which kinds a given machine +// will actually accept. +type WorkloadKind int + +const ( + // Compute is a batch job that runs and ends (today's Docker executor). + Compute WorkloadKind = iota + // Print shares a local printer (CUPS/USB) for one job. + Print + // Storage holds opaque sealed shards and proves possession on demand. + Storage + // Service is a long-running, pinned workload (a hosted "outcome") that + // stays up rather than completing. It requires Pinned placement. + Service +) + +func (k WorkloadKind) String() string { + switch k { + case Compute: + return "compute" + case Print: + return "print" + case Storage: + return "storage" + case Service: + return "service" + default: + return "unknown" + } +} + +// Valid reports whether k is one of the defined kinds. +func (k WorkloadKind) Valid() bool { return k >= Compute && k <= Service } + +// Placement is the capacity contract between the owner and the network. +type Placement int + +const ( + // Scavenged capacity is lent while idle and reclaimed the instant the + // owner returns. Only preemptible executors may run here. + Scavenged Placement = iota + // Pinned capacity is dedicated: declared, not scavenged. Services require + // it because they cannot yield to the owner mid-flight. + Pinned +) + +func (p Placement) String() string { + switch p { + case Scavenged: + return "scavenged" + case Pinned: + return "pinned" + default: + return "unknown" + } +} + +// AssignedJob is one accepted unit of work. Payload is opaque to the seam; the +// executor for the job's Kind interprets it. Deadline is advisory (zero means +// none) — enforcement is the executor's, using ctx. +type AssignedJob struct { + ID string + Kind WorkloadKind + Payload []byte + Deadline time.Time +} + +// Result is the outcome of running an AssignedJob. OK false with a nil error +// is a clean negative (e.g. a storage proof that did not verify); a non-nil +// error from Run is an operational failure to run at all. +type Result struct { + JobID string + OK bool + Detail string + Output []byte +} + +// Executor runs one accepted one-shot assignment on local resources. This is +// the seam every batch-style contributor implements (Docker compute, printing, +// storage store/audit). Long-running services implement ServiceExecutor +// instead. +type Executor interface { + // Kind is the single workload kind this executor serves. It must not be + // Service (services implement ServiceExecutor). + Kind() WorkloadKind + // Preemptible reports whether an in-flight run can be paused or stopped + // when the owner returns. Scavenged nodes require this; the Registry + // refuses a non-preemptible executor on a scavenged contract. + Preemptible() bool + // Run executes the job. It must honor ctx cancellation (that is how the + // owner-never-interrupted guarantee reaches a running scavenged job). + Run(ctx context.Context, job AssignedJob) (Result, error) +} + +// ServiceExecutor runs a long-running, pinned service. Serve blocks for the +// life of the service; Healthy is a liveness signal the agent can attest. +type ServiceExecutor interface { + // Kind must return Service. + Kind() WorkloadKind + // Healthy reports current liveness (for heartbeat attestation). + Healthy() bool + // Serve runs the service until ctx is cancelled or it fails fatally. + // A clean shutdown on ctx cancellation returns ctx.Err(). + Serve(ctx context.Context, job AssignedJob) error +} diff --git a/internal/contribute/registry.go b/internal/contribute/registry.go new file mode 100644 index 0000000..26c352f --- /dev/null +++ b/internal/contribute/registry.go @@ -0,0 +1,122 @@ +package contribute + +import ( + "context" + "errors" + "fmt" + "sort" +) + +var ( + // ErrKindNotContracted means an executor or job kind is outside the node + // contract. + ErrKindNotContracted = errors.New("contribute: workload kind not in node contract") + // ErrScavengedNotPreemptible means a non-preemptible executor was offered + // to a scavenged node, which would break the owner-never-interrupted + // guarantee. + ErrScavengedNotPreemptible = errors.New("contribute: scavenged node requires a preemptible executor") + // ErrDuplicateExecutor means a kind already has a registered executor. + ErrDuplicateExecutor = errors.New("contribute: executor already registered for kind") + // ErrUseRegisterService means a Service-kind executor was passed to + // Register instead of RegisterService. + ErrUseRegisterService = errors.New("contribute: service kind must be registered with RegisterService") + // ErrNotService means a ServiceExecutor did not report Kind()==Service. + ErrNotService = errors.New("contribute: service executor must report Kind()==Service") + // ErrNoExecutor means no executor is registered for a dispatched kind. + ErrNoExecutor = errors.New("contribute: no executor registered for kind") +) + +// Registry binds a node's contract to the concrete executors that satisfy it. +// It is the one place the contract's rules (contracted kinds only; scavenged +// requires preemptible; service requires pinned) are enforced at registration +// time, so a misconfiguration fails at startup rather than mid-assignment. +type Registry struct { + contract NodeContract + batch map[WorkloadKind]Executor + service ServiceExecutor +} + +// NewRegistry validates the contract and returns an empty registry for it. +func NewRegistry(c NodeContract) (*Registry, error) { + if err := c.Validate(); err != nil { + return nil, err + } + return &Registry{contract: c, batch: make(map[WorkloadKind]Executor)}, nil +} + +// Register adds a one-shot executor. It rejects the Service kind (use +// RegisterService), kinds outside the contract, non-preemptible executors on +// scavenged capacity, and a second executor for a kind already served. +func (r *Registry) Register(e Executor) error { + k := e.Kind() + if k == Service { + return ErrUseRegisterService + } + if !k.Valid() { + return fmt.Errorf("contribute: invalid kind %d", int(k)) + } + if !r.contract.Allows(k) { + return fmt.Errorf("%w: %s", ErrKindNotContracted, k) + } + if r.contract.Placement == Scavenged && !e.Preemptible() { + return fmt.Errorf("%w: %s", ErrScavengedNotPreemptible, k) + } + if _, dup := r.batch[k]; dup { + return fmt.Errorf("%w: %s", ErrDuplicateExecutor, k) + } + r.batch[k] = e + return nil +} + +// RegisterService adds the node's long-running service. The contract must +// permit Service (which, by NodeContract.Validate, guarantees Pinned +// placement), and only one service may be registered. +func (r *Registry) RegisterService(s ServiceExecutor) error { + if s.Kind() != Service { + return ErrNotService + } + if !r.contract.Allows(Service) { + return fmt.Errorf("%w: %s", ErrKindNotContracted, Service) + } + if r.service != nil { + return fmt.Errorf("%w: %s", ErrDuplicateExecutor, Service) + } + r.service = s + return nil +} + +// Capabilities returns the sorted kinds actually registered (and therefore +// permitted by the contract). This is what the agent advertises upstream — +// never more than the node truly serves. +func (r *Registry) Capabilities() []WorkloadKind { + out := make([]WorkloadKind, 0, len(r.batch)+1) + for k := range r.batch { + out = append(out, k) + } + if r.service != nil { + out = append(out, Service) + } + sort.Slice(out, func(i, j int) bool { return out[i] < out[j] }) + return out +} + +// Dispatch routes a one-shot job to its executor. It refuses jobs outside the +// contract or with no registered executor before any work runs. +func (r *Registry) Dispatch(ctx context.Context, job AssignedJob) (Result, error) { + if !r.contract.Allows(job.Kind) { + return Result{JobID: job.ID}, fmt.Errorf("%w: %s", ErrKindNotContracted, job.Kind) + } + e, ok := r.batch[job.Kind] + if !ok { + return Result{JobID: job.ID}, fmt.Errorf("%w: %s", ErrNoExecutor, job.Kind) + } + return e.Run(ctx, job) +} + +// Service returns the registered service executor, if any. +func (r *Registry) Service() (ServiceExecutor, bool) { + return r.service, r.service != nil +} + +// Contract returns the node contract this registry enforces. +func (r *Registry) Contract() NodeContract { return r.contract } diff --git a/internal/contribute/registry_test.go b/internal/contribute/registry_test.go new file mode 100644 index 0000000..f664e40 --- /dev/null +++ b/internal/contribute/registry_test.go @@ -0,0 +1,163 @@ +package contribute + +import ( + "context" + "errors" + "reflect" + "testing" +) + +// --- test doubles --- + +type fakeExec struct { + kind WorkloadKind + preempt bool + run func(context.Context, AssignedJob) (Result, error) +} + +func (f fakeExec) Kind() WorkloadKind { return f.kind } +func (f fakeExec) Preemptible() bool { return f.preempt } +func (f fakeExec) Run(ctx context.Context, j AssignedJob) (Result, error) { + if f.run != nil { + return f.run(ctx, j) + } + return Result{JobID: j.ID, OK: true}, nil +} + +type fakeSvc struct { + healthy bool + serve func(context.Context, AssignedJob) error +} + +func (f fakeSvc) Kind() WorkloadKind { return Service } +func (f fakeSvc) Healthy() bool { return f.healthy } +func (f fakeSvc) Serve(ctx context.Context, j AssignedJob) error { + if f.serve != nil { + return f.serve(ctx, j) + } + <-ctx.Done() + return ctx.Err() +} + +// --- tests --- + +func mustRegistry(t *testing.T, c NodeContract) *Registry { + t.Helper() + r, err := NewRegistry(c) + if err != nil { + t.Fatalf("NewRegistry: %v", err) + } + return r +} + +func TestNewRegistryRejectsBadContract(t *testing.T) { + if _, err := NewRegistry(NodeContract{Placement: Scavenged, Kinds: []WorkloadKind{Service}}); !errors.Is(err, ErrServiceNeedsPinned) { + t.Fatalf("want ErrServiceNeedsPinned, got %v", err) + } +} + +func TestRegisterContractAndPreemption(t *testing.T) { + // Scavenged node must reject a non-preemptible executor. + sc := mustRegistry(t, NodeContract{Placement: Scavenged, Kinds: []WorkloadKind{Compute}}) + if err := sc.Register(fakeExec{kind: Compute, preempt: false}); !errors.Is(err, ErrScavengedNotPreemptible) { + t.Fatalf("scavenged + non-preemptible: want ErrScavengedNotPreemptible, got %v", err) + } + if err := sc.Register(fakeExec{kind: Compute, preempt: true}); err != nil { + t.Fatalf("scavenged + preemptible should register: %v", err) + } + // Duplicate. + if err := sc.Register(fakeExec{kind: Compute, preempt: true}); !errors.Is(err, ErrDuplicateExecutor) { + t.Fatalf("want ErrDuplicateExecutor, got %v", err) + } + // Uncontracted kind. + if err := sc.Register(fakeExec{kind: Storage, preempt: true}); !errors.Is(err, ErrKindNotContracted) { + t.Fatalf("want ErrKindNotContracted, got %v", err) + } + // Service via Register is the wrong door. + if err := sc.Register(fakeExec{kind: Service, preempt: true}); !errors.Is(err, ErrUseRegisterService) { + t.Fatalf("want ErrUseRegisterService, got %v", err) + } + // Pinned node accepts a non-preemptible executor (dedicated capacity). + pin := mustRegistry(t, NodeContract{Placement: Pinned, Kinds: []WorkloadKind{Compute}}) + if err := pin.Register(fakeExec{kind: Compute, preempt: false}); err != nil { + t.Fatalf("pinned + non-preemptible should register: %v", err) + } +} + +func TestRegisterService(t *testing.T) { + pin := mustRegistry(t, NodeContract{Placement: Pinned, Kinds: []WorkloadKind{Service}}) + if err := pin.RegisterService(fakeSvc{}); err != nil { + t.Fatalf("RegisterService: %v", err) + } + if err := pin.RegisterService(fakeSvc{}); !errors.Is(err, ErrDuplicateExecutor) { + t.Fatalf("second service: want ErrDuplicateExecutor, got %v", err) + } + // A registry whose contract omits Service refuses one. + noSvc := mustRegistry(t, NodeContract{Placement: Pinned, Kinds: []WorkloadKind{Compute}}) + if err := noSvc.RegisterService(fakeSvc{}); !errors.Is(err, ErrKindNotContracted) { + t.Fatalf("want ErrKindNotContracted, got %v", err) + } +} + +func TestCapabilitiesSortedAndScoped(t *testing.T) { + r := mustRegistry(t, NodeContract{Placement: Pinned, Kinds: []WorkloadKind{Storage, Compute, Service}}) + if got := r.Capabilities(); len(got) != 0 { + t.Fatalf("empty registry should advertise nothing, got %v", got) + } + if err := r.Register(fakeExec{kind: Storage, preempt: true}); err != nil { + t.Fatal(err) + } + if err := r.Register(fakeExec{kind: Compute, preempt: true}); err != nil { + t.Fatal(err) + } + if err := r.RegisterService(fakeSvc{}); err != nil { + t.Fatal(err) + } + want := []WorkloadKind{Compute, Storage, Service} // sorted by iota order + if got := r.Capabilities(); !reflect.DeepEqual(got, want) { + t.Fatalf("Capabilities() = %v, want %v", got, want) + } +} + +func TestDispatch(t *testing.T) { + r := mustRegistry(t, NodeContract{Placement: Pinned, Kinds: []WorkloadKind{Compute, Storage}}) + if err := r.Register(fakeExec{kind: Compute, preempt: true, run: func(_ context.Context, j AssignedJob) (Result, error) { + return Result{JobID: j.ID, OK: true, Output: j.Payload}, nil + }}); err != nil { + t.Fatal(err) + } + + // Routes to the right executor and returns its result. + res, err := r.Dispatch(context.Background(), AssignedJob{ID: "j1", Kind: Compute, Payload: []byte("hi")}) + if err != nil || !res.OK || string(res.Output) != "hi" { + t.Fatalf("dispatch compute: res=%+v err=%v", res, err) + } + + // Contracted kind but no executor registered. + if _, err := r.Dispatch(context.Background(), AssignedJob{ID: "j2", Kind: Storage}); !errors.Is(err, ErrNoExecutor) { + t.Fatalf("want ErrNoExecutor, got %v", err) + } + + // Uncontracted kind refused before any executor runs. + if _, err := r.Dispatch(context.Background(), AssignedJob{ID: "j3", Kind: Print}); !errors.Is(err, ErrKindNotContracted) { + t.Fatalf("want ErrKindNotContracted, got %v", err) + } +} + +func TestServiceServeHonorsContext(t *testing.T) { + r := mustRegistry(t, NodeContract{Placement: Pinned, Kinds: []WorkloadKind{Service}}) + if err := r.RegisterService(fakeSvc{}); err != nil { + t.Fatal(err) + } + svc, ok := r.Service() + if !ok { + t.Fatal("Service() should report the registered service") + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- svc.Serve(ctx, AssignedJob{ID: "svc", Kind: Service}) }() + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("Serve should return context.Canceled, got %v", err) + } +}