diff --git a/docker-compose.yml b/docker-compose.yml index 2097f94a..fa53e50e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -213,7 +213,7 @@ services: start_period: 40s # PostgreSQL delivery store for the Go Worker. Profile-gated on `worker` - # for the same reason worker-store-init is: a stock deployment runs on + # because it is only ever an opt-in: a stock deployment runs on # SQLite and must not gain a database it never asked for. An operator opts # the durable queue onto PostgreSQL by setting ATF_DATABASE_URL on the # worker (and the Core, which owns the schema bootstrap). @@ -244,15 +244,25 @@ services: retries: 20 profiles: ["worker"] - # One-time preparation of the shared delivery store for the non-root - # worker. The Core creates atf.db as root; the worker (uid 100) must be - # able to write its claim updates to the SAME file. This runs only under - # the `worker` profile (i.e. when the operator opts into the Go Worker), - # so a stock deployment never sees it. + # Preparation of the shared delivery store for the non-root worker. + # The Core creates atf.db as root; the worker (uid 100) must be able to + # write its claim updates to the SAME file. This runs on EVERY `compose + # up` — the worker container itself is not profile-gated (it always runs + # to report health; only execution is gated by ATF_WORKER_ENABLED), so a + # stock deployment needs this too. It waits for the core to be healthy + # because on a fresh volume atf.db does not exist until the core creates + # it: an init that ran earlier would chgrp an empty directory and leave + # the newly created file root-owned — the exact crash loop + # ("probe write: attempt to write a readonly database") that failed the + # v2.5.0 clean-install gate. Idempotent: a second run is a no-op, so + # re-running after a Core restart is safe. worker-store-init: image: alpine:3.22 container_name: atf-worker-store-init restart: "no" + depends_on: + core: + condition: service_healthy volumes: - atf-data:/data # chgrp to the worker's gid (101 in the alpine image, from `addgroup -S`) @@ -261,7 +271,6 @@ services: command: > sh -c "chgrp -R 101 /data && chmod -R g+w /data || true; echo 'shared store prepared for the atf group'" - profiles: ["worker"] # Go execution engine (business execution plane). OFF by default: # ATF_WORKER_ENABLED must be set (AND ATF_CORE_RPC_TOKEN must be non-empty) @@ -344,6 +353,11 @@ services: depends_on: core: condition: service_healthy + # The store init must have chgrp'd/chmod'd atf.db before this + # container's probe write — without the wait the worker fail-closes + # on a root-owned database and restart-loops forever. + worker-store-init: + condition: service_completed_successfully # `required: false` is what makes the profile gate work: when the # `worker` profile is off, postgres does not exist and compose skips # this dependency instead of failing; when the profile is on, the diff --git a/worker/cmd/worker/main.go b/worker/cmd/worker/main.go index 2610594f..b43e0b62 100644 --- a/worker/cmd/worker/main.go +++ b/worker/cmd/worker/main.go @@ -53,6 +53,7 @@ import ( "atf/worker/internal/telegraph" "atf/worker/internal/telegraph/fakegateway" "atf/worker/internal/telegraph/grpcclient" + "atf/worker/internal/telegraph/unconfigured" ) // ContractVersion is the Telegram Gateway contract this build speaks. It @@ -174,10 +175,25 @@ func run() (runErr error) { default: client, err := grpcclient.New(cfg.TelegramRuntimeAddr, cfg.CoreRPCToken, workerID(), 60*time.Second) if err != nil { - return err + if !cfg.WorkerEnabled { + // The documented contract: an empty token refuses to + // start only an ENABLED worker. A disabled one exists to + // report health — the strangler-fig guarantee that the + // Python Core owns execution until the operator moves the + // boundary — so it boots with the health-only gateway + // instead of restart-looping with a dark /ops endpoint + // (the crash loop the v2.5.0 clean-install gate caught). + log.Warn("Telegram runtime client unconfigured — running health-only; enable the worker only after configuring it", + "error", err, + "telegram_runtime", cfg.TelegramRuntimeAddr) + gateway = unconfigured.New(err.Error()) + } else { + return err + } + } else { + defer client.Close() + gateway = client } - defer client.Close() - gateway = client } rules, err := openRuleStore(context.Background(), cfg) diff --git a/worker/internal/telegraph/unconfigured/unconfigured.go b/worker/internal/telegraph/unconfigured/unconfigured.go new file mode 100644 index 00000000..ce367bf0 --- /dev/null +++ b/worker/internal/telegraph/unconfigured/unconfigured.go @@ -0,0 +1,168 @@ +// Package unconfigured is the health-only telegraph.Gateway of a worker +// that cannot reach the Telegram runtime. +// +// The worker container always runs — even before the operator opts into Go +// execution — so it can report health. A worker in that state has no usable +// Telegram transport: ATF_CORE_RPC_TOKEN is empty, so the real gRPC client +// refuses to construct ("refusing an unauthenticated client"). Rather than +// refusing to START — which would leave the ops endpoint dark, the API's +// health payload degraded, and the container restart-looping (the crash +// loop the v2.5.0 clean-install gate caught) — the process boots with this +// gateway: every operation fails closed with a failure that names the +// missing configuration and the operation that was never attempted. +// +// The delivery engine never reaches it while ATF_WORKER_ENABLED is off. +// The moment the operator enables the worker, an unconfigured token is +// fatal at startup again — fail closed, on purpose. +// +// Like the fake gateway, this is never a substitute for a real delivery: +// nothing sent through it happened. +package unconfigured + +import ( + "context" + + "atf/worker/internal/telegraph" +) + +// Code is the canonical failure category for an operation that was refused +// because the worker has no Telegram transport configured. Not in +// DeliveryFailure.IsTransient: a missing configuration does not heal by +// retrying. +const Code = "TRANSPORT_UNCONFIGURED" + +// Gateway is the telegraph.Gateway of a health-only worker. +type Gateway struct { + // reason is the startup error the real client refused with, surfaced + // verbatim so the operator reads the actual cause, not a paraphrase. + reason string +} + +// New returns a Gateway whose every operation fails with reason. +func New(reason string) *Gateway { + return &Gateway{reason: reason} +} + +// refusal builds the failure every operation returns. The operation name is +// spelled out so a log line or Log Center row answers "what did NOT happen" +// without reading the stack. +func (g *Gateway) refusal(op string) *telegraph.DeliveryFailure { + return &telegraph.DeliveryFailure{ + Code: Code, + Message: "worker runs health-only: Telegram transport unconfigured (" + + g.reason + ") — " + op + " never attempted", + } +} + +// IsSessionReady reports the transport as never ready: a connecting session +// is not ready either, so callers defer rather than fail the job. +func (g *Gateway) IsSessionReady(_ context.Context, _ string) (bool, string, error) { + return false, "transport unconfigured (" + g.reason + ")", nil +} + +// --- reads ------------------------------------------------------------- // + +// GetMessage is refused: see the package comment. +func (g *Gateway) GetMessage(_ context.Context, _ string, _, _ int64) (*telegraph.MessageSnapshot, *telegraph.DeliveryFailure) { + return nil, g.refusal("GetMessage") +} + +// GetAlbumMessageIds is refused: see the package comment. +func (g *Gateway) GetAlbumMessageIds(_ context.Context, _ string, _, _ int64) ([]int64, *telegraph.DeliveryFailure) { + return nil, g.refusal("GetAlbumMessageIds") +} + +// GetFileExtension is refused: see the package comment. +func (g *Gateway) GetFileExtension(_ context.Context, _ string, _, _ int64) (string, *telegraph.DeliveryFailure) { + return "", g.refusal("GetFileExtension") +} + +// --- media staging ------------------------------------------------------ // + +// DownloadMedia is refused: see the package comment. +func (g *Gateway) DownloadMedia(_ context.Context, _ string, _, _ int64, _ telegraph.MediaHandle) (int64, *telegraph.DeliveryFailure) { + return 0, g.refusal("DownloadMedia") +} + +// ReloadMedia is refused: see the package comment. +func (g *Gateway) ReloadMedia(_ context.Context, _ string, _, _ int64, _ telegraph.MediaHandle, _ string) (int64, *telegraph.DeliveryFailure) { + return 0, g.refusal("ReloadMedia") +} + +// PrepareMediaUpload is refused: see the package comment. +func (g *Gateway) PrepareMediaUpload(_ context.Context, _ string, _ telegraph.MediaHandle, _, _ int64, _ string) (string, *telegraph.DeliveryFailure) { + return "", g.refusal("PrepareMediaUpload") +} + +// --- sends (the Telegram side effects) ---------------------------------- // + +// ForwardMessages is refused: see the package comment. +func (g *Gateway) ForwardMessages(_ context.Context, _ string, _, _ int64, _ []int64, _ int64) ([]int64, *telegraph.DeliveryFailure) { + return nil, g.refusal("ForwardMessages") +} + +// CopyMessage is refused: see the package comment. +func (g *Gateway) CopyMessage(_ context.Context, _ string, _, _, _ int64, _ string, _ int64, _ bool) (int64, *telegraph.DeliveryFailure) { + return 0, g.refusal("CopyMessage") +} + +// CopyMessages is refused: see the package comment. +func (g *Gateway) CopyMessages(_ context.Context, _ string, _, _ int64, _ []int64, _ string, _ int64, _ bool) ([]int64, *telegraph.DeliveryFailure) { + return nil, g.refusal("CopyMessages") +} + +// CopyMediaGroup is refused: see the package comment. +func (g *Gateway) CopyMediaGroup(_ context.Context, _ string, _, _ int64, _ int64, _ string, _ int64) ([]int64, *telegraph.DeliveryFailure) { + return nil, g.refusal("CopyMediaGroup") +} + +// SendMessage is refused: see the package comment. +func (g *Gateway) SendMessage(_ context.Context, _ string, _ int64, _ string, _ int64) (int64, *telegraph.DeliveryFailure) { + return 0, g.refusal("SendMessage") +} + +// SendMedia is refused: see the package comment. +func (g *Gateway) SendMedia(_ context.Context, _ string, _ int64, _, _, _ string, _ int64) (int64, *telegraph.DeliveryFailure) { + return 0, g.refusal("SendMedia") +} + +// --- bot transport ------------------------------------------------------ // + +// BotAvailability is refused: see the package comment. +func (g *Gateway) BotAvailability(_ context.Context, _ telegraph.BotTarget) (bool, string, *telegraph.DeliveryFailure) { + return false, "", g.refusal("BotAvailability") +} + +// BotSendText is refused: see the package comment. +func (g *Gateway) BotSendText(_ context.Context, _ telegraph.BotTarget, _ int64, _ string, _ int64, _ []string) (int64, *telegraph.DeliveryFailure) { + return 0, g.refusal("BotSendText") +} + +// BotCopyMessage is refused: see the package comment. +func (g *Gateway) BotCopyMessage(_ context.Context, _ telegraph.BotTarget, _, _, _ int64, _ string, _ int64, _ []string) (int64, *telegraph.DeliveryFailure) { + return 0, g.refusal("BotCopyMessage") +} + +// BotSendMedia is refused: see the package comment. +func (g *Gateway) BotSendMedia(_ context.Context, _ telegraph.BotTarget, _ int64, _, _, _ string, _ int64, _ []string) (int64, *telegraph.DeliveryFailure) { + return 0, g.refusal("BotSendMedia") +} + +// BotDeleteMessages is refused: see the package comment. +func (g *Gateway) BotDeleteMessages(_ context.Context, _ telegraph.BotTarget, _ int64, _ []int64) (bool, *telegraph.DeliveryFailure) { + return false, g.refusal("BotDeleteMessages") +} + +// --- deletion (auto-delete scheduler) ----------------------------------- // + +// DeleteMessages is refused: see the package comment. +func (g *Gateway) DeleteMessages(_ context.Context, _ telegraph.DeleteTransport, _, _ string, _, _ int64, _ []int64) (bool, *telegraph.DeliveryFailure) { + return false, g.refusal("DeleteMessages") +} + +// --- branding stage 1 ---------------------------------------------------- // + +// BrandingStage1 is refused: see the package comment. +func (g *Gateway) BrandingStage1(_ context.Context, _ telegraph.BrandingPost) ([]int64, *telegraph.DeliveryFailure) { + return nil, g.refusal("BrandingStage1") +} diff --git a/worker/internal/telegraph/unconfigured/unconfigured_test.go b/worker/internal/telegraph/unconfigured/unconfigured_test.go new file mode 100644 index 00000000..81d719b8 --- /dev/null +++ b/worker/internal/telegraph/unconfigured/unconfigured_test.go @@ -0,0 +1,101 @@ +package unconfigured + +import ( + "context" + "strings" + "testing" + + "atf/worker/internal/telegraph" +) + +// A compile-time shape check: the health-only gateway is a telegraph.Gateway, +// so it can drop into cmd/worker's disabled path. +var _ telegraph.Gateway = (*Gateway)(nil) + +// The health-only gateway must fail EVERY operation with the canonical +// non-transient code and a message that names the missing configuration — +// never a nil failure (which reads as "the operation happened") and never a +// panic (which would take the ops endpoint down with it). +func TestEveryOperationFailsClosed(t *testing.T) { + g := New("internal RPC token is empty: refusing an unauthenticated client") + ctx := context.Background() + target := telegraph.BotTarget{TokenID: "t1", OwnerUserID: 7} + + if ok, reason, err := g.IsSessionReady(ctx, "s1"); ok || err != nil || !strings.Contains(reason, "unconfigured") { + t.Fatalf("IsSessionReady = (%v, %q, %v), want (false, unconfigured reason, nil)", ok, reason, err) + } + + _, f := g.GetMessage(ctx, "s1", 1, 2) + check(t, "GetMessage", f, g.reason) + _, f = g.GetAlbumMessageIds(ctx, "s1", 1, 2) + check(t, "GetAlbumMessageIds", f, g.reason) + _, f = g.GetFileExtension(ctx, "s1", 1, 2) + check(t, "GetFileExtension", f, g.reason) + _, f = g.DownloadMedia(ctx, "s1", 1, 2, telegraph.MediaHandle{}) + check(t, "DownloadMedia", f, g.reason) + _, f = g.ReloadMedia(ctx, "s1", 1, 2, telegraph.MediaHandle{}, "photo") + check(t, "ReloadMedia", f, g.reason) + _, f = g.PrepareMediaUpload(ctx, "s1", telegraph.MediaHandle{}, 1, 2, "photo") + check(t, "PrepareMediaUpload", f, g.reason) + _, f = g.ForwardMessages(ctx, "s1", 1, 2, []int64{3}, 0) + check(t, "ForwardMessages", f, g.reason) + _, f = g.CopyMessage(ctx, "s1", 1, 2, 3, "", 0, false) + check(t, "CopyMessage", f, g.reason) + _, f = g.CopyMessages(ctx, "s1", 1, 2, []int64{3}, "", 0, false) + check(t, "CopyMessages", f, g.reason) + _, f = g.CopyMediaGroup(ctx, "s1", 1, 2, 3, "", 0) + check(t, "CopyMediaGroup", f, g.reason) + _, f = g.SendMessage(ctx, "s1", 1, "text", 0) + check(t, "SendMessage", f, g.reason) + _, f = g.SendMedia(ctx, "s1", 1, "tok", "cap", "photo", 0) + check(t, "SendMedia", f, g.reason) + _, f = g.BotSendText(ctx, target, 1, "text", 0, nil) + check(t, "BotSendText", f, g.reason) + _, f = g.BotCopyMessage(ctx, target, 1, 2, 3, "", 0, nil) + check(t, "BotCopyMessage", f, g.reason) + _, f = g.BotSendMedia(ctx, target, 1, "tok", "cap", "photo", 0, nil) + check(t, "BotSendMedia", f, g.reason) + _, f = g.BrandingStage1(ctx, telegraph.BrandingPost{}) + check(t, "BrandingStage1", f, g.reason) + + ok, username, f := g.BotAvailability(ctx, target) + if ok || username != "" { + t.Fatalf("BotAvailability = (%v, %q), want (false, \"\")", ok, username) + } + check(t, "BotAvailability", f, g.reason) + + ok, f = g.BotDeleteMessages(ctx, target, 1, []int64{2}) + if ok { + t.Fatalf("BotDeleteMessages = %v, want false", ok) + } + check(t, "BotDeleteMessages", f, g.reason) + + ok, f = g.DeleteMessages(ctx, telegraph.TransportUser, "s1", "b1", 7, 1, []int64{2}) + if ok { + t.Fatalf("DeleteMessages = %v, want false", ok) + } + check(t, "DeleteMessages", f, g.reason) +} + +// check asserts one refusal. +func check(t *testing.T, op string, f *telegraph.DeliveryFailure, reason string) { + t.Helper() + if f == nil { + t.Fatalf("%s returned nil failure — a nil reads as \"the operation happened\"", op) + } + if f.Code != Code { + t.Errorf("%s failure code = %q, want %q (message: %s)", op, f.Code, Code, f.Message) + } + if f.IsTransient() { + t.Errorf("%s: %q must not be transient — a missing configuration does not heal by retrying", op, f.Code) + } + if !strings.Contains(f.Message, "health-only") { + t.Errorf("%s refusal must name the health-only state, got: %s", op, f.Message) + } + if !strings.Contains(f.Message, reason) { + t.Errorf("%s refusal must carry the startup reason verbatim, got: %s", op, f.Message) + } + if !strings.Contains(f.Message, op) { + t.Errorf("%s refusal must name the operation that was never attempted, got: %s", op, f.Message) + } +}