From 9e54aba2b130de04dd684508b98186b1ea4d2749 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ois=C3=ADn=20Kyne?= Date: Mon, 6 Jul 2026 10:59:36 +0100 Subject: [PATCH] feat(buy): harden selling and buying to maximise successes --- cmd/obol/sell.go | 86 ++++++++++++++++++++++++++++++--- cmd/obol/sell_agent.go | 1 + cmd/obol/sell_test.go | 78 +++++++++++++++++++++++++++--- internal/x402/clockskew.go | 66 +++++++++++++++++++++++++ internal/x402/clockskew_test.go | 70 +++++++++++++++++++++++++++ 5 files changed, 289 insertions(+), 12 deletions(-) create mode 100644 internal/x402/clockskew.go create mode 100644 internal/x402/clockskew_test.go diff --git a/cmd/obol/sell.go b/cmd/obol/sell.go index ffc742b5..28f0559a 100644 --- a/cmd/obol/sell.go +++ b/cmd/obol/sell.go @@ -240,6 +240,7 @@ Examples: }, Action: func(ctx context.Context, cmd *cli.Command) error { u := getUI(cmd) + warnOnClockSkew(ctx, u, cmd.String("facilitator")) name := cmd.Args().First() if name == "" { if u.IsTTY() { @@ -802,6 +803,7 @@ Examples: }, acceptFlags()...), Action: func(ctx context.Context, cmd *cli.Command) error { u := getUI(cmd) + warnOnClockSkew(ctx, u, "") // --from-json: read spec from file/stdin and apply directly. if jsonPath := cmd.String("from-json"); jsonPath != "" { @@ -1435,6 +1437,11 @@ func serviceOfferStatusLines(namespace, name string, offer monetizeapi.ServiceOf for _, cond := range offer.Status.Conditions { lines = append(lines, formatConditionLine(cond)) } + if len(offer.Status.Conditions) > 0 { + lines = append(lines, "", + "Note: conditions are the controller's last reconciled snapshot — they update on", + "state changes, not per paid request. The chain is the canonical revenue record.") + } return lines } @@ -1539,9 +1546,29 @@ func formatConditionLine(cond monetizeapi.Condition) string { if cond.Message != "" { header = header + " — " + cond.Message } + if !cond.LastTransitionTime.IsZero() { + header = header + fmt.Sprintf(" (changed %s)", humanAge(cond.LastTransitionTime.Time)) + } return fmt.Sprintf(" %s %s", icon, header) } +// humanAge renders how long ago t was in the coarsest useful unit. Used +// on condition lines so sellers can tell a fresh reconcile from a status +// the controller last touched hours ago. +func humanAge(t time.Time) string { + d := time.Since(t) + switch { + case d < time.Minute: + return "just now" + case d < time.Hour: + return fmt.Sprintf("%dm ago", int(d.Minutes())) + case d < 24*time.Hour: + return fmt.Sprintf("%dh ago", int(d.Hours())) + default: + return fmt.Sprintf("%dd ago", int(d.Hours()/24)) + } +} + // conditionIcon picks a glyph based on the condition's status + reason. The // glyphs are plain unicode (no lipgloss) so the function is safe to call from // pure unit tests; coloring is applied at print time via the ui package. @@ -3941,6 +3968,7 @@ Examples: if err := waitForClusterAPI(ctx, cfg, u, 3*time.Minute); err != nil { u.Warnf("cluster API not ready: %v (continuing — per-offer applies may fail)", err) } + warnOnClockSkew(ctx, u, "") // Recorded Agent CRs first: agent-backed offers resolve // agent.ref and would dangle on a freshly-recreated cluster. agentcrd.ResumeAll(cfg, u) @@ -4085,6 +4113,35 @@ func installResumeBootUnit(cfg *config.Config, u *ui.UI) error { return nil } +// warnOnClockSkew is a best-effort seller preflight. EIP-3009 payment +// authorizations anchor validAfter/validBefore to wall clocks, so a +// drifted seller host makes the facilitator reject every buyer payment +// ("authorization is not yet valid") while `sell status` still looks +// healthy — revenue silently stops. Probes the facilitator's HTTP Date +// header and warns above the threshold; probe failures stay silent +// because the check must never block or delay selling. +func warnOnClockSkew(ctx context.Context, u *ui.UI, facilitatorURL string) { + if facilitatorURL == "" { + facilitatorURL = x402verifier.DefaultFacilitatorURL + } + + skew, err := x402verifier.MeasureClockSkew(ctx, facilitatorURL) + if err != nil { + return + } + if skew.Abs() <= x402verifier.ClockSkewWarnThreshold { + return + } + + direction := "ahead of" + if skew < 0 { + direction = "behind" + } + u.Warnf("This machine's clock is %s %s the payment facilitator (%s). Buyer payment authorizations may be rejected (\"authorization is not yet valid\") and revenue silently stops.", + skew.Abs().Round(time.Second), direction, facilitatorURL) + u.Dim(" Fix: enable NTP time sync (macOS: Date & Time → set automatically; Linux: timedatectl set-ntp true).") +} + // resumeSellOffers re-applies the cluster-side artifacts (Service + // Endpoints + ServiceOffer) for every locally-persisted `obol sell // inference` deployment, and relaunches each offer's host gateway as a @@ -4127,6 +4184,8 @@ func resumeSellOffers(ctx context.Context, cfg *config.Config, u *ui.UI) error { } deployments := activeInferenceDeployments(all) + var failed []string + if len(deployments) > 0 { u.Blank() u.Infof("Resuming %d locally-persisted sell-inference offer(s)...", len(deployments)) @@ -4135,6 +4194,7 @@ func resumeSellOffers(ctx context.Context, cfg *config.Config, u *ui.UI) error { for _, d := range deployments { if err := resumeOneInferenceOffer(cfg, u, d); err != nil { u.Warnf("resume %s: %v", d.Name, err) + failed = append(failed, d.Name) continue } resumed++ @@ -4151,9 +4211,22 @@ func resumeSellOffers(ctx context.Context, cfg *config.Config, u *ui.UI) error { // The two stores stay separate on disk because the inference // descriptor is rich (listen addr, asset, registration, gateway // PID) while the ledger holds only rendered ServiceOffer manifests. - if err := resumePersistedServiceOffers(cfg, u); err != nil { + persistedFailed, err := resumePersistedServiceOffers(cfg, u) + if err != nil { u.Warnf("resume persisted sell offers: %v", err) } + failed = append(failed, persistedFailed...) + + // A seller who reboots and sees "stack up: OK" assumes their offers + // are live. Per-offer warnings scroll past; this summary — and a + // non-nil error that fails `obol sell resume` outright — makes a + // partial resume impossible to miss. + if len(failed) > 0 { + u.Blank() + u.Errorf("%d sell offer(s) failed to resume and are NOT live for buyers: %s", len(failed), strings.Join(failed, ", ")) + u.Dim(" Retry with `obol sell resume`; inspect with `obol sell status -n `.") + return fmt.Errorf("%d sell offer(s) failed to resume: %s", len(failed), strings.Join(failed, ", ")) + } return nil } @@ -4861,18 +4934,18 @@ func removePersistedServiceOffersInNamespace(cfg *config.Config, ns string) (int // the cluster yet). Counts and announces what was reattached so the // operator sees the same "Resumed N offers" feedback they get for // inference. -func resumePersistedServiceOffers(cfg *config.Config, u *ui.UI) error { +func resumePersistedServiceOffers(cfg *config.Config, u *ui.UI) (failed []string, err error) { kubeconfigPath := filepath.Join(cfg.ConfigDir, "kubeconfig.yaml") if _, statErr := os.Stat(kubeconfigPath); statErr != nil { - return nil // no cluster yet + return nil, nil // no cluster yet } manifests, err := loadPersistedServiceOffers(sellOfferStoreDir(cfg), u) if err != nil { - return err + return nil, err } if len(manifests) == 0 { - return nil + return nil, nil } u.Blank() @@ -4880,11 +4953,12 @@ func resumePersistedServiceOffers(cfg *config.Config, u *ui.UI) error { for _, m := range manifests { if err := kubectlApply(cfg, m.Manifest); err != nil { u.Warnf("resume %s %s/%s: %v", m.label(), m.Namespace, m.Name, err) + failed = append(failed, fmt.Sprintf("%s/%s", m.Namespace, m.Name)) continue } u.Successf("Resumed %s offer %s/%s", m.label(), m.Namespace, m.Name) } - return nil + return failed, nil } // loadPersistedServiceOffers walks a ledger dir and parses every diff --git a/cmd/obol/sell_agent.go b/cmd/obol/sell_agent.go index 31b5d93a..c7ee9038 100644 --- a/cmd/obol/sell_agent.go +++ b/cmd/obol/sell_agent.go @@ -87,6 +87,7 @@ Examples: }, acceptFlags()...), Action: func(ctx context.Context, cmd *cli.Command) error { u := getUI(cmd) + warnOnClockSkew(ctx, u, "") if cmd.NArg() != 1 { return fmt.Errorf("agent name required: obol sell agent ") } diff --git a/cmd/obol/sell_test.go b/cmd/obol/sell_test.go index a06a716e..353cb728 100644 --- a/cmd/obol/sell_test.go +++ b/cmd/obol/sell_test.go @@ -8,6 +8,9 @@ import ( "runtime" "strings" "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "github.com/ObolNetwork/obol-stack/internal/config" "github.com/ObolNetwork/obol-stack/internal/inference" @@ -415,6 +418,60 @@ func TestServiceOfferStatusLines_RawTxFallback(t *testing.T) { } } +func TestFormatConditionLine_ShowsAge(t *testing.T) { + cond := monetizeapi.Condition{ + Type: "Ready", + Status: "True", + Reason: "Ready", + LastTransitionTime: metav1.Time{Time: time.Now().Add(-3 * time.Hour)}, + } + line := formatConditionLine(cond) + if !strings.Contains(line, "(changed 3h ago)") { + t.Errorf("condition line missing age, got: %q", line) + } + + // Zero transition time (older controllers, hand-built fixtures) must + // not render a bogus age. + cond.LastTransitionTime = metav1.Time{} + if line := formatConditionLine(cond); strings.Contains(line, "changed") { + t.Errorf("zero LastTransitionTime must not render an age, got: %q", line) + } +} + +func TestServiceOfferStatusLines_SnapshotCaveat(t *testing.T) { + withConditions := monetizeapi.ServiceOffer{ + Status: monetizeapi.ServiceOfferStatus{ + Conditions: []monetizeapi.Condition{{Type: "Ready", Status: "True"}}, + }, + } + joined := strings.Join(serviceOfferStatusLines("llm", "demo", withConditions, ""), "\n") + if !strings.Contains(joined, "last reconciled snapshot") { + t.Errorf("status output missing the snapshot caveat:\n%s", joined) + } + + // No conditions (offer not yet reconciled) → no caveat about them. + joined = strings.Join(serviceOfferStatusLines("llm", "demo", monetizeapi.ServiceOffer{}, ""), "\n") + if strings.Contains(joined, "snapshot") { + t.Errorf("caveat must only render when conditions exist:\n%s", joined) + } +} + +func TestHumanAge(t *testing.T) { + for _, tt := range []struct { + age time.Duration + want string + }{ + {30 * time.Second, "just now"}, + {5 * time.Minute, "5m ago"}, + {2 * time.Hour, "2h ago"}, + {49 * time.Hour, "2d ago"}, + } { + if got := humanAge(time.Now().Add(-tt.age)); got != tt.want { + t.Errorf("humanAge(-%v) = %q, want %q", tt.age, got, tt.want) + } + } +} + func TestFormatOfferAsset(t *testing.T) { tests := []struct { name string @@ -1834,9 +1891,13 @@ func TestResumeSellHTTPOffers_EmptyStoreNoOp(t *testing.T) { if err := os.WriteFile(filepath.Join(cfg.ConfigDir, "kubeconfig.yaml"), []byte("placeholder"), 0o600); err != nil { t.Fatalf("seed kubeconfig: %v", err) } - if err := resumePersistedServiceOffers(cfg, ui.New(false)); err != nil { + failed, err := resumePersistedServiceOffers(cfg, ui.New(false)) + if err != nil { t.Errorf("empty-store resume must succeed: %v", err) } + if len(failed) != 0 { + t.Errorf("empty-store resume must report no failed offers, got %v", failed) + } } // TestSellDeleteAction_CallsRemoveSellHTTPOffer is a source-level guard @@ -1885,11 +1946,16 @@ func TestResumeSellOffers_HTTPOnlyStore(t *testing.T) { if err := persistServiceOffer(cfg, "llm", "only-http", manifest); err != nil { t.Fatalf("persist: %v", err) } - // kubectl apply will fail against the fake kubeconfig — that's - // expected and reported as a warn, not an error. We just need - // resumeSellOffers itself to return nil. - if err := resumeSellOffers(context.Background(), cfg, ui.New(false)); err != nil { - t.Errorf("http-only store must not error from resumeSellOffers: %v", err) + // kubectl apply fails against the fake kubeconfig, which means the + // offer did NOT come back — resumeSellOffers must say so loudly: a + // seller who reboots and reads "OK" while their offer is down loses + // revenue silently. The error must name the failed offer. + err := resumeSellOffers(context.Background(), cfg, ui.New(false)) + if err == nil { + t.Fatal("resumeSellOffers must return an error when an offer fails to resume") + } + if !strings.Contains(err.Error(), "llm/only-http") { + t.Errorf("resume error must name the failed offer, got: %v", err) } } diff --git a/internal/x402/clockskew.go b/internal/x402/clockskew.go new file mode 100644 index 00000000..243e9156 --- /dev/null +++ b/internal/x402/clockskew.go @@ -0,0 +1,66 @@ +package x402 + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "time" +) + +// ClockSkewWarnThreshold is the local-vs-facilitator clock offset above +// which sellers should be warned. EIP-3009 authorizations are signed with +// validAfter/validBefore anchored to the signer's wall clock; a drifted +// host makes the facilitator (and USDC's FiatTokenV2 contract) reject +// otherwise-valid payments with "authorization is not yet valid" — +// revenue silently stops while everything looks healthy. The HTTP Date +// header has 1s resolution and rides normal request latency, so the +// threshold stays well above measurement noise. +const ClockSkewWarnThreshold = 30 * time.Second + +// clockSkewProbeTimeout bounds the skew probe so preflight checks never +// hold up a sell command on a slow network. +const clockSkewProbeTimeout = 3 * time.Second + +// MeasureClockSkew measures the local clock's offset from the +// facilitator's clock via the HTTP Date response header: positive means +// the local clock is ahead. The request midpoint is used as the local +// reference to cancel out request latency. Returns an error when the +// facilitator is unreachable or sends no Date header — callers treat the +// probe as best-effort. +func MeasureClockSkew(ctx context.Context, facilitatorURL string) (time.Duration, error) { + ctx, cancel := context.WithTimeout(ctx, clockSkewProbeTimeout) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, facilitatorURL, nil) + if err != nil { + return 0, fmt.Errorf("build skew probe: %w", err) + } + + start := time.Now() + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return 0, fmt.Errorf("skew probe: %w", err) + } + defer resp.Body.Close() + + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1024)) + + elapsed := time.Since(start) + + dateHeader := resp.Header.Get("Date") + if dateHeader == "" { + return 0, errors.New("facilitator response has no Date header") + } + + serverTime, err := http.ParseTime(dateHeader) + if err != nil { + return 0, fmt.Errorf("parse facilitator Date header: %w", err) + } + + localMidpoint := start.Add(elapsed / 2) + + return localMidpoint.Sub(serverTime), nil +} diff --git a/internal/x402/clockskew_test.go b/internal/x402/clockskew_test.go new file mode 100644 index 00000000..29483407 --- /dev/null +++ b/internal/x402/clockskew_test.go @@ -0,0 +1,70 @@ +package x402 + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestMeasureClockSkew_DetectsOffset(t *testing.T) { + // Server reports a Date two minutes in the past → local clock appears + // two minutes ahead (positive skew). + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Date", time.Now().Add(-2*time.Minute).UTC().Format(http.TimeFormat)) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + skew, err := MeasureClockSkew(context.Background(), srv.URL) + if err != nil { + t.Fatalf("MeasureClockSkew: %v", err) + } + + // Date-header resolution is 1s and the probe adds latency; a ±5s + // tolerance keeps this robust on slow CI machines. + if skew < 2*time.Minute-5*time.Second || skew > 2*time.Minute+5*time.Second { + t.Errorf("skew = %v, want ~2m", skew) + } +} + +func TestMeasureClockSkew_InSync(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Date", time.Now().UTC().Format(http.TimeFormat)) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + skew, err := MeasureClockSkew(context.Background(), srv.URL) + if err != nil { + t.Fatalf("MeasureClockSkew: %v", err) + } + + if skew.Abs() > 5*time.Second { + t.Errorf("skew = %v, want ~0 for an in-sync server", skew) + } + if skew.Abs() > ClockSkewWarnThreshold { + t.Errorf("in-sync server must not cross the warn threshold, got %v", skew) + } +} + +func TestMeasureClockSkew_NoDateHeader(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // httptest's default server writes a Date header automatically; + // suppress it to simulate a facilitator that omits one. + w.Header()["Date"] = nil + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + if _, err := MeasureClockSkew(context.Background(), srv.URL); err == nil { + t.Error("expected error when the response has no Date header") + } +} + +func TestMeasureClockSkew_Unreachable(t *testing.T) { + if _, err := MeasureClockSkew(context.Background(), "http://127.0.0.1:1"); err == nil { + t.Error("expected error for an unreachable facilitator") + } +}