provider services take part in the image phase, with image distribution on demand

docs/examples/provider.go

@@ -18,9 +18,12 @@ package main
 
 import (
 	"bufio"
+	"crypto/sha256"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net"
+	"net/http/httputil"
 	"os"
 	"os/exec"
 	"strings"
@@ -85,11 +88,88 @@ func composeCommand() *cobra.Command {
 		Args: cobra.ExactArgs(1),
 	}
 
-	c.AddCommand(upCmd, downCmd, stopCmd)
-	c.AddCommand(metadataCommand(upCmd, downCmd, stopCmd))
+	// pull is the image-distribution command: compose invokes it during the
+	// image phase (up) and on `docker compose pull`, passing the identity of
+	// the service image and the state of the local daemon cache. None of its
+	// flags is required: they are injected by compose, not declared by the
+	// user under provider.options.
+	pullCmd := &cobra.Command{
+		Use:  "pull",
+		Run:  pull,
+		Args: cobra.ExactArgs(1),
+	}
+	pullCmd.Flags().String("image", "", "Image reference as compose resolved it")
+	pullCmd.Flags().String("digest", "", "Image ID in the local daemon cache, when present")
+	pullCmd.Flags().String("created", "", "Creation time of the local cache entry, when present")
+	pullCmd.Flags().String("source", "", "Authority verdict: local (sync from the daemon) or registry (resolve upstream)")
+	pullCmd.Flags().String("policy", "", "missing (a usable version suffices) or always (ensure freshness)")
+
+	c.AddCommand(upCmd, downCmd, stopCmd, pullCmd)
+	c.AddCommand(metadataCommand(upCmd, downCmd, stopCmd, pullCmd))
 	return c
 }
 
+// pull demonstrates the image-distribution contract. A real provider would
+// compare the announced digest/created with its bookkeeping and either
+// resolve the reference upstream (source=registry) or request the local
+// bytes (source=local). This demo requests the stream whenever
+// PROVIDER_PULL_MARKER is set, hashes it, and records the outcome there.
+func pull(cmd *cobra.Command, args []string) {
+	image, _ := cmd.Flags().GetString("image")
+	digest, _ := cmd.Flags().GetString("digest")
+	created, _ := cmd.Flags().GetString("created")
+	source, _ := cmd.Flags().GetString("source")
+	policy, _ := cmd.Flags().GetString("policy")
+
+	emit := func(kind, message string) {
+		payload, _ := json.Marshal(map[string]string{"type": kind, "message": message})
+		fmt.Println(string(payload))
+	}
+	emit("info", fmt.Sprintf("ensuring image %s (source=%s, policy=%s)", image, source, policy))
+
+	marker := os.Getenv("PROVIDER_PULL_MARKER")
+	if marker == "" {
+		// nothing to synchronize in the demo: a real registry-sourced provider
+		// would pull the reference upstream here
+		return
+	}
+
+	// Request the image bytes from the local daemon. The answer is one JSON
+	// line, then — unless it carries an error — the tar as an HTTP/1.1
+	// chunked body (readable with any stock chunked reader), whose zero
+	// chunk marks a COMPLETE transfer.
+	request, _ := json.Marshal(map[string]string{"type": "get-image", "message": image})
+	fmt.Println(string(request))
+
+	stdin := bufio.NewReader(os.Stdin)
+	line, err := stdin.ReadString('\n')
+	if err != nil {
+		emit("error", "reading image-stream announce: "+err.Error())
+		os.Exit(1)
+	}
+	var announce struct {
+		Encoding string `json:"encoding"`
+		Error    string `json:"error"`
+	}
+	if err := json.Unmarshal([]byte(line), &announce); err != nil || announce.Error != "" || announce.Encoding != "chunked" {
+		emit("error", fmt.Sprintf("image stream unavailable: %q (err %v)", line, err))
+		os.Exit(1)
+	}
+	h := sha256.New()
+	n, err := io.Copy(h, httputil.NewChunkedReader(stdin))
+	if err != nil {
+		emit("error", "image stream truncated: "+err.Error())
+		os.Exit(1)
+	}
+	record := fmt.Sprintf("image=%s digest=%s created=%s source=%s policy=%s sha256=%x bytes=%d\n",
+		image, digest, created, source, policy, h.Sum(nil), n)
+	if err := os.WriteFile(marker, []byte(record), 0o600); err != nil {
+		emit("error", "writing marker: "+err.Error())
+		os.Exit(1)
+	}
+	emit("info", fmt.Sprintf("image received (%d bytes)", n))
+}
+
 // 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
@@ -227,23 +307,27 @@ func stop(_ *cobra.Command, _ []string) {
 	}
 }
 
-func metadataCommand(upCmd, downCmd, stopCmd *cobra.Command) *cobra.Command {
+func metadataCommand(upCmd, downCmd, stopCmd, pullCmd *cobra.Command) *cobra.Command {
 	return &cobra.Command{
 		Use: "metadata",
 		Run: func(cmd *cobra.Command, _ []string) {
-			metadata(upCmd, downCmd, stopCmd)
+			metadata(upCmd, downCmd, stopCmd, pullCmd)
 		},
 		Args: cobra.NoArgs,
 	}
 }
 
-func metadata(upCmd, downCmd, stopCmd *cobra.Command) {
+func metadata(upCmd, downCmd, stopCmd, pullCmd *cobra.Command) {
 	metadata := ProviderMetadata{}
 	metadata.Description = "Manage services on AwesomeCloud"
 	metadata.Up = commandParameters(upCmd)
 	metadata.Down = commandParameters(downCmd)
 	stopParams := commandParameters(stopCmd)
 	metadata.Stop = &stopParams
+	// like stop, declaring the pull block is what opts the provider into
+	// image distribution
+	pullParams := commandParameters(pullCmd)
+	metadata.Pull = &pullParams
 	jsonMetadata, err := json.Marshal(metadata)
 	if err != nil {
 		panic(err)
@@ -271,6 +355,7 @@ type ProviderMetadata struct {
 	Up          CommandMetadata  `json:"up"`
 	Down        CommandMetadata  `json:"down"`
 	Stop        *CommandMetadata `json:"stop,omitempty"`
+	Pull        *CommandMetadata `json:"pull,omitempty"`
 }
 
 type CommandMetadata struct {

docs/extension.md

@@ -30,7 +30,10 @@ the resource(s) needed to run a service.
 If `provider.type` doesn't resolve into any of those, Compose will report an error and interrupt the `up` command.
 
 To be a valid Compose extension, provider command *MUST* accept a `compose` command (which can be hidden)
-with subcommands `up` and `down`. It *MAY* additionally implement a `stop` subcommand to support `docker compose stop`.
+with subcommands `up` and `down`. It *MAY* additionally implement a `stop` subcommand to support `docker compose stop`,
+and a `pull` subcommand to take part in image distribution (see [Image distribution](#image-distribution)).
+Optional subcommands are declared through the provider metadata: the presence of the command block is what
+opts the provider in.
 
 ## Up lifecycle
 
@@ -111,6 +114,68 @@ sequenceDiagram
     Compose-)Shell: service started
 ```
 
+## Image distribution
+
+A provider-backed service can declare `build` or `image` like any other service; the image then has to reach
+the provider's runtime, which may be nowhere near the local daemon. Providers opt into image distribution by
+declaring a `pull` block in their `metadata` output — like `stop`, the presence of the block is the
+declaration of support. Providers without it keep managing images on their own during `up`.
+
+When the provider declares `pull`, Compose invokes it during the image phase of `up` (after any build) and on
+`docker compose pull`:
+
+```console
+awesomecloud compose --project-name <NAME> pull --image=<ref> --source=<verdict> --policy=<policy> [--digest=<id> --created=<time>] "database"
+```
+
+- `--image`: the image reference as Compose resolved it (the `image` attribute, or `<project>-<service>` for a
+  build-only service).
+- `--digest` / `--created`: the state of the **local daemon cache**, present only when the image exists there.
+  They describe the cache, they are not instructions: persist them as the bookkeeping keys of what you ingest —
+  the digest as identity test, `created` as the ordering fallback for a backend that cannot preserve digests.
+  Beware that reproducible builds can freeze `created`, so a comparable digest always wins over it.
+- `--source` is the authority verdict, computed by Compose from the model and the invocation (`pull_policy`,
+  `--build`, what the current run just built), so providers never re-implement that arbitration:
+  - `local`: the desired state is the local daemon's image. Compare your bookkeeping with the announced
+    digest/created; when they differ, request the bytes with `get-image`.
+  - `registry`: resolve the reference upstream — this includes the common workflow where `build` is only the
+    recipe CI uses to publish the image that consumers pull. The local facts are an optimization, never an
+    obligation.
+- `--policy`:
+  - `missing` (the `up` path): a usable version present in your runtime suffices;
+  - `always` (`docker compose pull`): ensure your runtime holds the latest version of the authority.
+
+### Requesting the image bytes
+
+During `pull`, the provider can ask Compose for the image content with a regular JSON line on `stdout`
+(`platform` is optional and narrows a multi-platform image):
+
+```json
+{ "type": "get-image", "message": "<image ref>", "platform": "linux/arm64" }
+```
+
+Compose answers on the provider's `stdin` with one JSON line:
+
+```json
+{ "type": "image-stream", "encoding": "chunked", "media-type": "application/x-tar" }
+```
+
+followed — unless the line carries an `error` field instead — by the image tar encoded as HTTP/1.1 chunked
+data (RFC 9112 §7.1): every block of data prefixed by its length, terminated by the zero-length chunk. Unlike
+an HTTP message there is no trailer section nor final CRLF — the next byte after the zero chunk belongs to the
+next stdin answer. Length-prefixed framing needs no in-band delimiter (any byte value can appear inside a tar), and
+any language's stock chunked-body reader consumes it — Go providers can use `httputil.NewChunkedReader`. A
+stream that ends without the terminating zero chunk was aborted and must be discarded — Compose closes the
+answer channel after an aborted transfer, so the truncation is always observable as EOF. The tar is what
+`docker image save` produces: feed it to `docker load` or whatever your runtime ingests.
+
+As during `stop`, any `setenv`, `rawsetenv` or `publish-endpoint` message emitted during `pull` is accepted
+but ignored: dependent services are not being configured in this phase.
+
+The stream is exclusive on `stdin` for its whole duration — the chunked body must be contiguous, so answers to
+any other request emitted meanwhile are delivered after it. Drain the announced stream completely before
+expecting another answer.
+
 ## Connection to a service managed by a provider
 
 A service in the Compose application can declare dependency on a service managed by an external provider: 
@@ -269,6 +334,9 @@ The expected JSON output format is:
         "type": "string"
       }
     ]
+  },
+  "pull": {
+    "parameters": []
   }
 }
 ```
@@ -277,6 +345,9 @@ The top elements are:
 - `up`: Object describing the parameters accepted by the `up` command
 - `down`: Object describing the parameters accepted by the `down` command
 - `stop`: Object describing the parameters accepted by the `stop` command (optional)
+- `pull`: Object describing the parameters accepted by the `pull` command (optional — declaring the block is
+  what opts the provider into [image distribution](#image-distribution); the flags Compose injects need not be
+  listed)
 
 And for each command parameter, you should include the following properties:
 - `name`: The parameter name (without `--` prefix)

pkg/compose/build.go

@@ -106,7 +106,7 @@ func (s *composeService) build(ctx context.Context, project *types.Project, opti
 	return s.doBuildClassic(ctx, project, serviceToBuild, options)
 }
 
-func (s *composeService) ensureImagesExists(ctx context.Context, project *types.Project, buildOpts *api.BuildOptions, quietPull bool) error {
+func (s *composeService) ensureImagesExists(ctx context.Context, project *types.Project, buildOpts *api.BuildOptions, quietPull bool, skipProviders bool) error {
 	for name, service := range project.Services {
 		if service.Provider == nil && service.Image == "" && service.Build == nil {
 			return fmt.Errorf("invalid service %q. Must specify either image or build", name)
@@ -127,6 +127,7 @@ func (s *composeService) ensureImagesExists(ctx context.Context, project *types.
 		return err
 	}
 
+	built := map[string]string{}
 	if buildOpts != nil {
 		err = tracing.SpanWrapFunc("project/build", tracing.ProjectOptions(ctx, project),
 			func(ctx context.Context) error {
@@ -136,6 +137,7 @@ func (s *composeService) ensureImagesExists(ctx context.Context, project *types.
 				}
 
 				for name, digest := range builtImages {
+					built[name] = digest
 					images[name] = api.ImageSummary{
 						Repository:  name,
 						ID:          digest,
@@ -150,6 +152,20 @@ func (s *composeService) ensureImagesExists(ctx context.Context, project *types.
 		}
 	}
 
+	// provider-backed services get their turn in the image phase: the
+	// provider makes the image available to ITS runtime (weak contract on
+	// the up path: a usable version present suffices)
+	if !skipProviders {
+		err = tracing.SpanWrapFunc("project/provider-pull", tracing.ProjectOptions(ctx, project),
+			func(ctx context.Context) error {
+				return s.ensureProviderImages(ctx, project, built, providerPullPolicyMissing)
+			},
+		)(ctx)
+		if err != nil {
+			return err
+		}
+	}
+
 	// set digest as com.docker.compose.image label so we can detect outdated
 	// containers — the single writer of that label, so the platform-pinned
 	// resolution below can't be overwritten by another code path

pkg/compose/create.go

@@ -80,7 +80,7 @@ func (s *composeService) create(ctx context.Context, project *types.Project, opt
 		return err
 	}
 
-	err = s.ensureImagesExists(ctx, project, options.Build, options.QuietPull)
+	err = s.ensureImagesExists(ctx, project, options.Build, options.QuietPull, options.SkipProviders)
 	if err != nil {
 		return err
 	}

pkg/compose/images_test.go

@@ -423,7 +423,7 @@ func TestPlatformPinnedDigest(t *testing.T) {
 			Return(client.ImageInspectResult{InspectResponse: multiPlatform}, nil)
 
 		project := newProject()
-		assert.NilError(t, tested.ensureImagesExists(t.Context(), project, nil, true))
+		assert.NilError(t, tested.ensureImagesExists(t.Context(), project, nil, true, false))
 		assert.Equal(t, project.Services["app"].CustomLabels[compose.ImageDigestLabel], "sha256:s390x")
 	})