micro/go-micro · error

subscribe to %s: %w

Error message

subscribe to %s: %w

What it means

Flow registration failed because the broker subscription to the flow's trigger topic returned an error. The wrapped error (%w) carries the broker-specific cause — typically a broker not configured, connection failure, or invalid topic.

Source

Thrown at flow/flow.go:152

		modelOpts = append(modelOpts, ai.WithTools(f.toolSet))

		f.model = ai.New(f.opts.Provider, modelOpts...)
		if f.model == nil {
			return fmt.Errorf("unknown provider: %s", f.opts.Provider)
		}
	}

	if f.opts.TriggerTopic != "" {
		sub, err := br.Subscribe(f.opts.TriggerTopic, func(p broker.Event) error {
			data := string(p.Message().Body)
			ctx := ai.WithRunInfo(context.Background(), ai.RunInfo{Dispatch: "broker", Trigger: f.opts.TriggerTopic})
			if err := f.Execute(ctx, data); err != nil {
				f.log.Logf(logger.ErrorLevel, "Flow %s failed: %v", f.name, err)
			}
			return nil
		})
		if err != nil {
			return fmt.Errorf("subscribe to %s: %w", f.opts.TriggerTopic, err)
		}
		f.sub = sub
		f.log.Logf(logger.InfoLevel, "Flow %s subscribed to %s", f.name, f.opts.TriggerTopic)

		// Announce the flow in the registry so it's discoverable like a
		// service or agent (e.g. `micro flow list`). This is liveness only:
		// Stop deregisters it. Durable run history lives in the store.
		f.registration = &registry.Service{
			Name:    f.name,
			Version: "latest",
			Metadata: map[string]string{
				"type":    "flow",
				"trigger": f.opts.TriggerTopic,
				"steps":   strconv.Itoa(len(f.opts.Steps)),
			},
			Nodes: []*registry.Node{{
				Id:       f.name + "-" + uuid.New().String()[:8],
				Address:  "flow://" + f.name,

View on GitHub (pinned to 24529f1404)

Solutions

1. Ensure a broker is initialized and connected before Register (broker.Init / service Init).
2. Verify broker connectivity (broker.Connect) and inspect the wrapped error for dial failures.
3. Validate the TriggerTopic string (no spaces/illegal chars, non-empty).
4. Check broker-side ACLs/permissions if using an authenticated broker.

Example fix

// before
f.Register(ctx) // broker never initialized
// after
br.Init(broker.Backend(natsbroker.NewBroker(natsbroker.Addrs("nats://localhost:4222"))))
br.Connect(ctx)
f.Register(ctx)

if err := br.Connect(ctx); err != nil {
	return fmt.Errorf("broker not connected before flow Register: %w", err)
}

func brokerReady(br broker.Broker) bool { return br != nil && br.String() != "" }

if err := f.Register(ctx); err != nil {
	if strings.Contains(err.Error(), "subscribe to") {
		time.Sleep(2 * time.Second)
		return f.Register(ctx) // broker may still be connecting
	}
	return err
}

When it happens

Trigger: Calling Register on a flow with TriggerTopic set when br.Subscribe fails — e.g. no broker initialized in the micro service, broker connection down, or invalid/empty topic syntax.

Common situations: Forgetting micro broker init in standalone tests; NATS/Redis broker unreachable at startup; topic names with illegal characters; permissions rejecting the subscription.

Related errors

• API error: nil response
• cannot init while connected
• not connected
• ErrIPNotFound
• failed request

AI-assisted analysis of micro/go-micro@24529f1404 (2026-09-01). Data as JSON: /api/errors/3bc1aa7426cc3b73. Report an issue: GitHub.