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,52 @@ 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 subprocess deliberately outlives this invocation — it IS the
+		// provisioned resource the relay forwards to, and consumers connect
+		// through it only after up has returned, so reaping it here would
+		// tear the endpoint down before anyone reached it. Its lifetime is
+		// its own: it exits by itself after three minutes (serve-demo), the
+		// way a real provider's resource outlives the provider CLI run. No
+		// Wait() and no zombie either: this process exits within seconds, so
+		// the subprocess is long re-parented to init — which reaps it — when
+		// its three minutes are up.
+		//
+		// 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)
@@ -124,7 +213,12 @@ func up(options options, args []string) {
 }
 
 func down(_ *cobra.Command, _ []string) {
-	fmt.Printf(`{ "type": "error", "message": "Permission error" }%s`, lineSeparator)
+	// A failing down can be simulated for tests and demos.
+	if os.Getenv("PROVIDER_DOWN_FAILURE") != "" {
+		fmt.Printf(`{ "type": "error", "message": "Permission error" }%s`, lineSeparator)
+		return
+	}
+	fmt.Printf(`{ "type": "info", "message": "Resource removed" }%s`, lineSeparator)
 }
 
 func stop(_ *cobra.Command, _ []string) {

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/down.go

@@ -96,13 +96,7 @@ func (s *composeService) down(ctx context.Context, projectName string, options a
 	}
 
 	err = InReverseDependencyOrder(ctx, project, func(c context.Context, service string) error {
-		serv := project.Services[service]
-		if serv.Provider != nil {
-			return s.runPlugin(ctx, project, serv, "down")
-		}
-		serviceContainers := containers.filter(isService(service))
-		err := s.removeContainers(ctx, serviceContainers, &serv, options.Timeout, options.Volumes)
-		return err
+		return s.downService(ctx, project, containers, options, service)
 	}, WithRootNodesAndDown(options.Services))
 	if err != nil {
 		return err
@@ -384,6 +378,22 @@ func (s *composeService) stopAndRemoveContainer(ctx context.Context, ctr contain
 	return nil
 }
 
+// downService removes one service's containers. A provider service may still
+// own project containers — the relay deployed when it published endpoints —
+// and the plugin only removes the provider's own resource, so the containers
+// go first, mirroring up, which provisions the resource before the relay.
+func (s *composeService) downService(ctx context.Context, project *types.Project, containers Containers, options api.DownOptions, service string) error {
+	serv := project.Services[service]
+	serviceContainers := containers.filter(isService(service))
+	if err := s.removeContainers(ctx, serviceContainers, &serv, options.Timeout, options.Volumes); err != nil {
+		return err
+	}
+	if serv.Provider != nil {
+		return s.runPlugin(ctx, project, serv, "down")
+	}
+	return nil
+}
+
 func (s *composeService) getProjectWithResources(ctx context.Context, containers Containers, projectName string) (*types.Project, error) {
 	containers = containers.filter(isNotOneOff)
 	p, err := s.projectFromName(containers, projectName)

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)
 	}
 }