Artifact [9b7cfec09f]

Artifact 9b7cfec09f99e71c10480eaae7ae7501260198786bdafe4720c3e09d68211113:

• File transport.go — part of check-in [47dc6363c0] at 2026-10-12 19:01:46 on branch main — fix(sync): frame HTTP requests and replies as fossil's http.c does (#310) (#342)

  The server refused application/x-fossil-debug with HTTP 415, where fossil dispatches it to sync like the other x-fossil types. The client read any 2xx as a sync reply and decoded a reply as compressed only when typed exactly application/x-fossil, so a reply with no Content-Type was read as plain cards and a proxy's HTML page was parsed as cards.

  Replies now go through xfer.DecodeReply, which applies section 4's reply rule by case-insensitive prefix: no type is the compressed container, an x-fossil-debug or x-fossil-uncompressed prefix is card text, another x-fossil prefix is compressed, and anything else is refused. Only status 200 is a reply. (user: [email protected], size: 5146) [annotate] [blame] [check-ins using]

package libfossil

import (
	"context"
	"errors"
	"fmt"
	"net/http"

	internalsync "github.com/danmestas/go-libfossil/internal/sync"
	"github.com/danmestas/go-libfossil/internal/xfer"
)

// Transport delivers sync payloads between peers.
// Implementations handle the network layer (HTTP, NATS, etc.).
// Payloads are opaque zlib-compressed xfer card streams.
type Transport interface {
	RoundTrip(ctx context.Context, payload []byte) ([]byte, error)
}

// FramedTransport is an optional interface a Transport may also implement when
// it can report the framing signal — such as an HTTP Content-Type header — that
// accompanied a reply. Sync uses that signal to select the reply's §4 framing
// instead of assuming the §4.1 compressed container, which matters when the
// peer is a different Fossil-protocol implementation that legitimately replies
// with the uncompressed plain-card framing (e.g. a real Fossil server's clone
// reply).
//
// It is optional in the same way http.Flusher is optional on an
// http.ResponseWriter: a Transport whose framing is fixed by convention (both
// ends always emit the compressed container) need not implement it, and sync
// falls back to assuming the compressed framing for such a Transport. Callers
// type-assert a Transport to FramedTransport and use RoundTripFramed when it is
// available. Reporting the signal as a return value keeps it per-call, with no
// shared mutable state to make a reused Transport unsafe across concurrent
// calls.
type FramedTransport interface {
	Transport
	// RoundTripFramed behaves like RoundTrip but also returns the framing
	// signal (e.g. the reply's Content-Type) the wire carried for this reply,
	// or "" when the transport observed none.
	RoundTripFramed(ctx context.Context, payload []byte) (resp []byte, contentType string, err error)
}

// NewHTTPTransport creates a Transport that speaks Fossil's HTTP /xfer protocol.
func NewHTTPTransport(url string, opts ...HTTPOption) Transport {
	t := &httpTransport{url: url, client: http.DefaultClient}
	for _, o := range opts {
		o(t)
	}
	return t
}

// HTTPOption configures an HTTP transport.
type HTTPOption func(*httpTransport)

// WithHTTPClient sets a custom http.Client for the transport.
func WithHTTPClient(c *http.Client) HTTPOption {
	return func(t *httpTransport) { t.client = c }
}

type httpTransport struct {
	url    string
	client *http.Client
}

func (t *httpTransport) RoundTrip(ctx context.Context, payload []byte) ([]byte, error) {
	body, _, err := t.roundTrip(ctx, payload)
	return body, err
}

// RoundTripFramed implements FramedTransport: it returns the reply body along
// with the reply's Content-Type, which selects the §4 framing for decoding. A
// peer that replies uncompressed (e.g. a real Fossil clone reply) is decoded
// correctly instead of being force-read as the compressed container.
func (t *httpTransport) RoundTripFramed(ctx context.Context, payload []byte) ([]byte, string, error) {
	return t.roundTrip(ctx, payload)
}

func (t *httpTransport) roundTrip(ctx context.Context, payload []byte) ([]byte, string, error) {
	resp, err := internalsync.PostXfer(ctx, t.client, t.url, "application/x-fossil", payload)
	if err != nil {
		return nil, "", fmt.Errorf("libfossil: http transport: %w", err)
	}
	defer resp.Body.Close()
	// Only 200 is a sync reply (§4-7). Any other status, a 2xx included, is not
	// the protocol answering. Reporting the status directly keeps a peer's
	// outage from surfacing as a framing defect, and lets the sync round loop
	// tell a 5xx worth retrying from a 4xx that never will be.
	if resp.StatusCode != http.StatusOK {
		return nil, "", &internalsync.StatusError{Code: resp.StatusCode, Status: resp.Status}
	}
	body, err := xfer.ReadBody(resp.Body)
	if errors.Is(err, xfer.ErrBodyTooLarge) {
		// The reply is the peer's, and resending gets the same one.
		return nil, "", err
	}
	if err != nil {
		// The reply was cut short on the wire; nothing has been decoded yet,
		// so the round loop may safely send this request again. Tagging it
		// here is what distinguishes a severed connection from a body that
		// arrived whole and would not decode — both surface as
		// io.ErrUnexpectedEOF, and only the former is worth retrying.
		return nil, "", internalsync.RetryableFault(
			fmt.Errorf("libfossil: http transport read: %w", err))
	}
	return body, resp.Header.Get("Content-Type"), nil
}

// MockTransport is a test double that delegates to a handler function.
type MockTransport struct {
	Handler func(req []byte) []byte
}

// RoundTrip calls the Handler function if set, otherwise returns empty bytes.
func (t *MockTransport) RoundTrip(_ context.Context, payload []byte) ([]byte, error) {
	if t.Handler == nil {
		return []byte{}, nil
	}
	return t.Handler(payload), nil
}

// TransportFunc adapts a plain function to the Transport interface.
// This is the Transport equivalent of http.HandlerFunc.
type TransportFunc func(ctx context.Context, payload []byte) ([]byte, error)

// RoundTrip calls the function.
func (f TransportFunc) RoundTrip(ctx context.Context, payload []byte) ([]byte, error) {
	return f(ctx, payload)
}