feat(provider): publish-endpoint deploys a network relay for provider services

docker-bake.hcl

@@ -166,3 +166,12 @@ target "image-module-cross" {
     "windows/arm64",
   ]
 }
+
+target "relay-image" {
+  context = "./relay"
+  tags = ["docker/compose-relay:v1"]
+  platforms = [
+    "linux/amd64",
+    "linux/arm64",
+  ]
+}

docs/examples/provider.go

@@ -17,9 +17,13 @@
 package main
 
 import (
+	"bufio"
 	"encoding/json"
 	"fmt"
+	"net"
 	"os"
+	"os/exec"
+	"strings"
 	"time"
 
 	"github.com/spf13/cobra"
@@ -32,6 +36,7 @@ func main() {
 		Use:   "demo",
 	}
 	cmd.AddCommand(composeCommand())
+	cmd.AddCommand(serveDemoCommand())
 	err := cmd.Execute()
 	if err != nil {
 		fmt.Fprintln(os.Stderr, err)
@@ -85,6 +90,44 @@ func composeCommand() *cobra.Command {
 	return c
 }
 
+// serveDemoCommand is the detached helper process behind the
+// publish-endpoint demonstration: a TCP server on the given address
+// answering every connection with a fixed HTTP response, exiting on its own
+// after three minutes. It owns the port from bind to exit: the bound address
+// is reported on stdout once listening, so the parent never has to probe or
+// pre-reserve the port (no TOCTOU window, and works on Windows where handing
+// a socket over ExtraFiles is not supported).
+func serveDemoCommand() *cobra.Command {
+	return &cobra.Command{
+		Use:    "serve-demo ADDR",
+		Hidden: true,
+		Args:   cobra.ExactArgs(1),
+		RunE: func(_ *cobra.Command, args []string) error {
+			listener, err := net.Listen("tcp", args[0])
+			if err != nil {
+				return err
+			}
+			fmt.Println(listener.Addr().String())
+			go func() {
+				time.Sleep(3 * time.Minute)
+				os.Exit(0)
+			}()
+			for {
+				conn, err := listener.Accept()
+				if err != nil {
+					return err
+				}
+				go func() {
+					defer func() { _ = conn.Close() }()
+					buf := make([]byte, 1024)
+					_, _ = conn.Read(buf)
+					_, _ = conn.Write([]byte("HTTP/1.1 200 OK\r\nContent-Length: 19\r\nConnection: close\r\n\r\nhello from provider"))
+				}()
+			}
+		},
+	}
+}
+
 const lineSeparator = "\n"
 
 func up(options options, args []string) {
@@ -115,6 +158,42 @@ func up(options options, args []string) {
 	setenv, _ := json.Marshal(map[string]string{"type": "setenv", "message": "CONFIG_TYPE=" + config.Provider.Type})
 	fmt.Println(string(setenv))
 
+	// When asked to, stand up a real endpoint on the host and publish it, so
+	// compose deploys a relay and consumers reach it as http://<service>:80.
+	if os.Getenv("PROVIDER_DEMO_ENDPOINT") != "" {
+		// The subprocess binds the port itself and reports the resulting
+		// address on its stdout; only then is the endpoint published. This
+		// avoids the two races of a pre-reserved port: another process
+		// grabbing it between release and re-bind, and publish-endpoint
+		// pointing at a server that is not listening yet.
+		// All interfaces, not loopback: on a plain Linux engine host-gateway
+		// is the bridge IP, which cannot reach a host loopback bind.
+		server := exec.Command(os.Args[0], "serve-demo", "0.0.0.0:0")
+		stdout, err := server.StdoutPipe()
+		if err != nil {
+			fmt.Printf(`{ "type": "error", "message": "demo endpoint: %v" }%s`, err, lineSeparator)
+			return
+		}
+		if err := server.Start(); err != nil {
+			fmt.Printf(`{ "type": "error", "message": "demo endpoint: %v" }%s`, err, lineSeparator)
+			return
+		}
+		// A crashed subprocess closes the pipe (EOF below); a hung one would
+		// block the read forever, so kill it after a deadline — the read then
+		// fails with EOF and lands on the same error path.
+		watchdog := time.AfterFunc(30*time.Second, func() { _ = server.Process.Kill() })
+		addr, err := bufio.NewReader(stdout).ReadString('\n')
+		watchdog.Stop()
+		if err != nil {
+			fmt.Printf(`{ "type": "error", "message": "demo endpoint did not come up: %v" }%s`, err, lineSeparator)
+			return
+		}
+		// the endpoint is announced as seen from THIS process's host —
+		// the relay translates loopback into the container-visible name
+		_, port, _ := net.SplitHostPort(strings.TrimSpace(addr))
+		fmt.Printf(`{ "type": "publish-endpoint", "message": "80=localhost:%s" }%s`, port, lineSeparator)
+	}
+
 	for i := 0; i < options.size; i += 10 {
 		time.Sleep(1 * time.Second)
 		fmt.Printf(`{ "type": "info", "message": "Processing ... %d%%" }%s`, i*100/options.size, lineSeparator)

docs/extension.md

@@ -60,6 +60,25 @@ JSON messages MUST include a `type` and a `message` attribute.
 - `rawsetenv`: Same as `setenv`, but the variable is injected as-is without the service name prefix. Useful when applications require exact variable names that cannot be altered.
 - `debug`: Those messages could help debugging the provider, but are not rendered to the user by default. They are rendered when Compose is started with `--verbose` flag.
 - `get-service-config`: Asks Compose for the resolved configuration of the service the provider manages. See next section.
+- `publish-endpoint`: Declares where a network endpoint of the provider's resource is actually reachable. The
+  message is `"<container-port>=<host>:<port>"` — the port consumers know on the left, the real location on the
+  right, as seen FROM THE PROVIDER'S HOST (typically a port published on the host):
+  ```json
+  { "type": "publish-endpoint", "message": "80=localhost:49152" }
+  ```
+  The provider does not need to know how containers reach its host: the relay translates a loopback (or
+  unspecified) upstream host into `host.docker.internal` — resolved through the `host-gateway` extra_host
+  Compose injects — while routable addresses pass through untouched.
+  When a provider publishes at least one endpoint, Compose deploys a **relay container** in place of the service:
+  a minimal TCP forwarder (`docker/compose-relay` — set `COMPOSE_RELAY_IMAGE` to pull the image from an internal
+  registry instead of Docker Hub) joining the networks of the services that depend on the
+  provider service, aliased with the service name. Consumers then reach the resource at the compose-native
+  address — `http://<service>:<container-port>` — with no injected variables involved. The relay is a regular
+  project container (standard compose labels, canonical `<project>-<service>-1` name), so `ps`, `logs`, `stop`
+  and `down` treat it as the service; it additionally carries the `com.docker.compose.relay` label identifying
+  its role, and process-level commands (`exec`, `cp`) refuse it. The relay is recreated when the published
+  endpoints change, and removed by `down` like any project container. TCP only; the message may be repeated,
+  one per port.
 
 ## Requesting the service configuration
 

pkg/api/labels.go

@@ -43,6 +43,12 @@ const (
 	EnvironmentFileLabel = "com.docker.compose.project.environment_file"
 	// OneoffLabel stores value 'True' for one-off containers created by `compose run`
 	OneoffLabel = "com.docker.compose.oneoff"
+	// RelayLabel marks the network relay container compose deploys in place
+	// of a provider-managed service (see the publish-endpoint provider
+	// message). Its value is a hash of the relay's routes, used to decide
+	// whether an existing relay can be kept on the next up. Commands that
+	// act on a service's process (exec, ...) refuse relay containers.
+	RelayLabel = "com.docker.compose.relay"
 	// SlugLabel stores unique slug used for one-off container identity
 	SlugLabel = "com.docker.compose.slug"
 	// ImageDigestLabel stores digest of the container image used to run service

pkg/compose/cp.go

@@ -121,6 +121,9 @@ func (s *composeService) listContainersTargetedForCopy(ctx context.Context, proj
 		if err != nil {
 			return nil, err
 		}
+		if err := checkRelayTarget(ctr, serviceName, "cp"); err != nil {
+			return nil, err
+		}
 		return append(containers, ctr), nil
 	default:
 		withOneOff := oneOffExclude
@@ -131,6 +134,11 @@ func (s *composeService) listContainersTargetedForCopy(ctx context.Context, proj
 		if err != nil {
 			return nil, err
 		}
+		for _, ctr := range containers {
+			if err := checkRelayTarget(ctr, serviceName, "cp"); err != nil {
+				return nil, err
+			}
+		}
 
 		if len(containers) < 1 {
 			return nil, fmt.Errorf("no container found for service %q", serviceName)

pkg/compose/exec.go

@@ -33,6 +33,9 @@ func (s *composeService) Exec(ctx context.Context, projectName string, options a
 	if err != nil {
 		return 0, err
 	}
+	if err := checkRelayTarget(target, options.Service, "exec"); err != nil {
+		return 0, err
+	}
 
 	exec := container.NewExecOptions()
 	exec.Interactive = options.Interactive

pkg/compose/monitor.go

@@ -184,13 +184,21 @@ func (c *monitor) initialContainers(ctx context.Context) (utils.Set[string], err
 	}
 	containers := utils.Set[string]{}
 	for _, ctr := range initialState.Items {
-		if c.watched(ctr.Labels[api.ServiceLabel]) {
+		if c.watched(ctr.Labels[api.ServiceLabel]) && !isRelay(ctr.Labels) {
 			containers.Add(ctr.ID)
 		}
 	}
 	return containers, nil
 }
 
+// isRelay reports whether labels identify a provider-relay container. Relays
+// are long-lived infrastructure standing in for a provider's resource: they
+// never terminate on their own, so counting them among the application's
+// containers would keep an attached `up` waiting forever.
+func isRelay(labels map[string]string) bool {
+	return labels[api.RelayLabel] != ""
+}
+
 // watched tells whether a service's containers are watched by this monitor.
 // An empty service set means "the whole application".
 func (c *monitor) watched(service string) bool {
@@ -205,7 +213,7 @@ func (c *monitor) notify(event api.ContainerEvent) {
 }
 
 func (c *monitor) onContainerCreate(event events.Message, ctr *api.ContainerSummary, containers utils.Set[string]) {
-	if c.watched(ctr.Labels[api.ServiceLabel]) {
+	if c.watched(ctr.Labels[api.ServiceLabel]) && !isRelay(ctr.Labels) {
 		containers.Add(ctr.ID)
 	}
 	evtType := api.ContainerEventCreated
@@ -226,7 +234,7 @@ func (c *monitor) onContainerStart(event events.Message, ctr *api.ContainerSumma
 		logrus.Debugf("container %s started", ctr.Name)
 		c.notify(newContainerEvent(event.TimeNano, ctr, api.ContainerEventStarted))
 	}
-	if c.watched(ctr.Labels[api.ServiceLabel]) {
+	if c.watched(ctr.Labels[api.ServiceLabel]) && !isRelay(ctr.Labels) {
 		containers.Add(ctr.ID)
 	}
 }

pkg/compose/plugins.go

@@ -23,9 +23,11 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net"
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"sync"
 
@@ -50,6 +52,7 @@ const (
 	SetEnvType                = "setenv"
 	RawSetEnvType             = "rawsetenv"
 	DebugType                 = "debug"
+	PublishEndpointType       = "publish-endpoint"
 	providerMetadataDirectory = "compose/providers"
 
 	// GetServiceConfigType is a message the provider sends to receive, on
@@ -61,6 +64,11 @@ const (
 type pluginVariables struct {
 	prefixed types.Mapping
 	raw      types.Mapping
+	// endpoints are "port=host:port" publish-endpoint messages: container
+	// port the consumers know, mapped to where the provider's resource
+	// actually listens. When present, compose deploys a relay container
+	// under the service's name on the consumers' networks.
+	endpoints map[int]string
 }
 
 var mux sync.Mutex
@@ -107,9 +115,30 @@ func (s *composeService) runPlugin(ctx context.Context, project *types.Project,
 			project.Services[name] = s
 		}
 	}
+	if command == "up" && len(variables.endpoints) > 0 {
+		if err := s.ensureServiceRelay(ctx, project, service, variables.endpoints); err != nil {
+			return err
+		}
+	}
 	return nil
 }
 
+// parseEndpointMessage decodes a publish-endpoint payload: "80=host:port".
+func parseEndpointMessage(message string) (int, string, error) {
+	portPart, upstream, found := strings.Cut(message, "=")
+	if !found {
+		return 0, "", fmt.Errorf("publish-endpoint %q: want port=host:port", message)
+	}
+	port, err := strconv.Atoi(portPart)
+	if err != nil || port < 1 || port > 65535 {
+		return 0, "", fmt.Errorf("publish-endpoint %q: invalid port", message)
+	}
+	if _, _, err := net.SplitHostPort(upstream); err != nil {
+		return 0, "", fmt.Errorf("publish-endpoint %q: invalid endpoint: %w", message, err)
+	}
+	return port, upstream, nil
+}
+
 func (s *composeService) executePlugin(cmd *exec.Cmd, command string, service types.ServiceConfig) (pluginVariables, error) {
 	var action string
 	switch command {
@@ -180,8 +209,9 @@ func (s *composeService) executePlugin(cmd *exec.Cmd, command string, service ty
 	defer func() { _ = stdout.Close() }()
 
 	variables := pluginVariables{
-		prefixed: types.Mapping{},
-		raw:      types.Mapping{},
+		prefixed:  types.Mapping{},
+		raw:       types.Mapping{},
+		endpoints: map[int]string{},
 	}
 
 	for {
@@ -224,6 +254,12 @@ func (s *composeService) executePlugin(cmd *exec.Cmd, command string, service ty
 				defer stdinMu.Unlock()
 				_, _ = stdin.Write(payload)
 			}()
+		case PublishEndpointType:
+			port, upstream, err := parseEndpointMessage(msg.Message)
+			if err != nil {
+				return pluginVariables{}, fmt.Errorf("invalid response from plugin: %w", err)
+			}
+			variables.endpoints[port] = upstream
 		case DebugType:
 			logrus.Debugf("%s: %s", service.Name, msg.Message)
 		default: