Artifact [875c251e09]

Artifact 875c251e0933f6b72aed393325dc944b40d533d6faaa2d81303dc83eafe86b81:

• File hub/nats_transport.go — part of check-in [6f9e3e40d0] at 2026-05-12 20:27:38 on branch trunk — feat(hub): cross-leaf fossil sync over NATS (#156) (#157)

  Add Config.ProjectCode + Config.SeedFromUpstream so multiple hubs sharing a project-code propagate commits over a single .sync subject. Adds an auto-publish notification on a sibling .commit subject + pull-on-notify subscriber, closing the cross-process Fossil sharing gap from #156.

  - hub/nats_transport.go: libfossil.Transport over nats request/reply - hub/clone.go: HTTP-based SeedFromUpstream helper (NATS-clone deferred) - hub/repo.go: Repo.publish hook fired after successful Commit - hub/hub.go: startCommitSubscriber + Config plumbing + NoEcho - hub/cross_leaf_test.go: gold-path propagation, subject alignment, drift guard - libfossil pin -> v0.6.1 (CreateOpts.ProjectCode)

  Design: docs/superpowers/specs/2026-05-12-cross-leaf-fossil-sync-design.md Protocol: docs/architecture/sync-protocol.md ("NATS-Wrapped Sync")

  Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> (user: [email protected], size: 1754) [annotate] [blame] [check-ins using]

package hub

import (
	"context"
	"errors"
	"fmt"
	"time"

	libfossil "github.com/danmestas/libfossil"
	"github.com/nats-io/nats.go"
)

// natsTransport is a libfossil.Transport over NATS request/reply on a
// single subject. RoundTrip publishes the xfer payload as a request,
// waits for the responder, and returns the response bytes.
//
// Used by the hub to pull from peers on `<prefix>.<project-code>.sync`
// after receiving a notification on the sibling `.commit` subject.
type natsTransport struct {
	nc      *nats.Conn
	subject string
	timeout time.Duration
}

const defaultNATSRoundTripTimeout = 30 * time.Second

func newNATSTransport(nc *nats.Conn, subject string, timeout time.Duration) libfossil.Transport {
	if timeout <= 0 {
		timeout = defaultNATSRoundTripTimeout
	}
	return &natsTransport{nc: nc, subject: subject, timeout: timeout}
}

// RoundTrip publishes payload on the configured subject and blocks for
// the reply, honoring ctx. Returns an error if the responder takes
// longer than the transport's timeout (whichever fires first between
// ctx and timeout). An empty reply body is treated as an error — the
// hub-side handler uses an empty body to signal a HandleSync failure,
// and a libfossil Sync round needs a non-empty manifest to make
// progress.
func (t *natsTransport) RoundTrip(ctx context.Context, payload []byte) ([]byte, error) {
	reqCtx, cancel := context.WithTimeout(ctx, t.timeout)
	defer cancel()

	msg, err := t.nc.RequestWithContext(reqCtx, t.subject, payload)
	if err != nil {
		return nil, fmt.Errorf("nats transport: request %s: %w", t.subject, err)
	}
	if len(msg.Data) == 0 {
		return nil, errors.New("nats transport: empty reply (responder signaled HandleSync failure)")
	}
	return msg.Data, nil
}