Artifact [e77c1c2189]

Artifact e77c1c2189c29a3c848569c0ce352096b275648b9a41b1741e7752f2a9b955ee:

• File internal/sync/transport.go — part of check-in [b2928e9629] at 2026-07-27 20:03:00 on branch trunk — fix(sync): retry a dropped exchange round instead of aborting the clone (#185) (#196)

  A clone of a large repository is a conversation of dozens of rounds spanning minutes. A single transient disconnect anywhere in it aborted the whole operation with "http transport: Post ...: EOF" and produced no store at all: there was no retry, no reconnect, and minutes of transfer were discarded for one dropped packet. Reported failures landed at rounds 6, 11 and 18 across otherwise identical runs, which is the signature of independent random drops rather than a deterministic defect.

  Both round loops now send a failed round again after a bounded exponential backoff — four attempts, 250 ms doubling to a 2 s cap — but only when the failure was the network dropping the exchange. The classification is an allow-list, not a deny-list: a reply that arrived whole and would not decode, a card the peer rejected, a 4xx, or an unrecognised error from a Transport this package does not know all still fail on their first occurrence. Retrying those would turn a reproducible defect into an intermittent one, which is worse than the failure retrying exists to prevent.

  Telling the two apart cannot be done reliably after the fact — a truncated zlib body and a truncated wire read both surface as io.ErrUnexpectedEOF — so the transports tag their own wire stages via the optional TransportFault interface, in the same shape as FramedTransport. Non-2xx replies are now reported as such instead of being handed to the decoder, which also stops a peer's outage from being misreported as a framing defect.

  The caller's context governs throughout: a cancellation or deadline aborts immediately, including during a backoff wait, so retrying cannot extend an operation past the bound its caller set (#166, #167, #183).

  Resuming a partially-completed clone across process exit is deliberately not included: it needs the clone cursor persisted, a staleness policy for the server's snapshot bound, and a separate entry point, and within-session retry already preserves the expensive part. (user: [email protected], size: 3212) [annotate] [blame] [check-ins using]

package sync

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"

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

// Transport sends an xfer request and returns the response.
// Implementations handle encoding (the request Message is already decoded),
// network I/O, and decoding. The xfer.Message uses zlib-compressed payloads
// internally — see [xfer.Encode] and [xfer.Decode] for wire format details.
//
// Built-in implementations: [HTTPTransport] (Fossil HTTP /xfer protocol),
// [MockTransport] (canned responses for testing). The leaf agent adds
// NATSTransport for NATS-based peer-to-peer sync.
type Transport interface {
	Exchange(ctx context.Context, request *xfer.Message) (*xfer.Message, error)
}

// MockTransport replays canned responses for testing.
type MockTransport struct {
	Handler func(req *xfer.Message) *xfer.Message
}

func (t *MockTransport) Exchange(ctx context.Context, req *xfer.Message) (*xfer.Message, error) {
	if req == nil {
		panic("sync.MockTransport.Exchange: req must not be nil")
	}
	if t.Handler == nil {
		return &xfer.Message{}, nil
	}
	return t.Handler(req), nil
}

// HTTPTransport speaks Fossil's HTTP /xfer protocol.
// Fossil routes to /xfer based on Content-Type: application/x-fossil,
// NOT the URL path. URL should be the repo root (e.g. "http://localhost:8080").
type HTTPTransport struct {
	URL string // repo root, e.g. "http://localhost:8080"
}

func (t *HTTPTransport) Exchange(ctx context.Context, req *xfer.Message) (*xfer.Message, error) {
	if req == nil {
		panic("sync.HTTPTransport.Exchange: req must not be nil")
	}
	body, err := req.Encode()
	if err != nil {
		return nil, fmt.Errorf("sync.HTTPTransport encode: %w", err)
	}
	httpReq, err := http.NewRequestWithContext(ctx, "POST", t.URL, bytes.NewReader(body))
	if err != nil {
		return nil, fmt.Errorf("sync.HTTPTransport request: %w", err)
	}
	httpReq.Header.Set("Content-Type", xfer.ContentTypeCompressed)
	resp, err := http.DefaultClient.Do(httpReq)
	if err != nil {
		return nil, fmt.Errorf("sync.HTTPTransport do: %w", err)
	}
	defer resp.Body.Close()
	// An error page is not an xfer reply: report the status rather than
	// handing HTML to the decoder, which would report the peer's outage as a
	// framing defect. Whether the status is worth another attempt is
	// StatusError's to say.
	if resp.StatusCode < 200 || resp.StatusCode > 299 {
		return nil, &StatusError{Code: resp.StatusCode, Status: resp.Status}
	}
	respBody, err := io.ReadAll(resp.Body)
	if err != nil {
		// The reply was cut short on the wire. Nothing has been interpreted
		// yet, so the same request may safely be sent again.
		return nil, &transportFault{
			err:       fmt.Errorf("sync.HTTPTransport read: %w", err),
			retryable: true,
		}
	}
	// §4: the reply's framing is given by its Content-Type, not guessed. A
	// clone-v3 reply arrives uncompressed; a pull reply arrives as the
	// compressed container.
	//
	// A failure from here on is a protocol fault: the reply arrived whole and
	// did not parse. It is left untagged so the retry classifier fails it
	// fast — resending would only make a malformed reply intermittent.
	return xfer.Decode(respBody, resp.Header.Get("Content-Type"))
}