diff --git a/daemon/containerd/image_snapshot.go b/daemon/containerd/image_snapshot.go
index 24f2fd62b4ecf..cb128905c8bec 100644
--- a/daemon/containerd/image_snapshot.go
+++ b/daemon/containerd/image_snapshot.go
@@ -3,6 +3,8 @@ package containerd
 import (
 	"context"
 	"fmt"
+	"strconv"
+	"strings"
 
 	c8dimages "github.com/containerd/containerd/v2/core/images"
 	"github.com/containerd/containerd/v2/core/leases"
@@ -15,6 +17,7 @@ import (
 	"github.com/docker/docker/errdefs"
 	"github.com/docker/docker/image"
 	"github.com/docker/docker/layer"
+	"github.com/docker/go-units"
 	"github.com/opencontainers/image-spec/identity"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/pkg/errors"
@@ -44,6 +47,39 @@ func (i *ImageService) CreateLayerFromImage(img *image.Image, layerName string,
 	return i.createLayer(descriptor, layerName, rwLayerOpts, nil)
 }
 
+func (i *ImageService) generatePrepareOpts(ctx context.Context, rwLayerOpts *layer.CreateRWLayerOpts) ([]snapshots.Opt, error) {
+	var opts []snapshots.Opt
+
+	if rwLayerOpts != nil && len(rwLayerOpts.StorageOpt) > 0 {
+		for key, val := range rwLayerOpts.StorageOpt {
+			key = strings.ToLower(key)
+			switch key {
+			case "size":
+				size, err := units.RAMInBytes(val)
+				if err != nil {
+					return nil, err
+				}
+
+				if storageDriver := i.StorageDriver(); storageDriver == "windows" {
+					sizeLabel := map[string]string{
+						"containerd.io/snapshot/windows/rootfs.sizebytes": strconv.FormatInt(size, 10),
+					}
+					opts = append(opts, snapshots.WithLabels(sizeLabel))
+				} else {
+					/// TODO: containerd doesn't handle quotas for most snapshotters
+					/// See: https://github.com/containerd/containerd/issues/759 and related
+					log.G(ctx).Warnf("--storage-opt is not supported for the %s driver", storageDriver)
+				}
+
+			default:
+				return nil, fmt.Errorf("Unknown option %s", key)
+			}
+		}
+	}
+
+	return opts, nil
+}
+
 func (i *ImageService) createLayer(descriptor *ocispec.Descriptor, layerName string, rwLayerOpts *layer.CreateRWLayerOpts, initFunc layer.MountInit) (container.RWLayer, error) {
 	ctx := context.TODO()
 	var parentSnapshot string
@@ -78,7 +114,13 @@ func (i *ImageService) createLayer(descriptor *ocispec.Descriptor, layerName str
 	if !i.idMapping.Empty() {
 		err = i.remapSnapshot(ctx, sn, layerName, parentSnapshot)
 	} else {
-		_, err = sn.Prepare(ctx, layerName, parentSnapshot)
+		var sopts []snapshots.Opt
+		sopts, err = i.generatePrepareOpts(ctx, rwLayerOpts)
+		if err != nil {
+			return nil, err
+		}
+
+		_, err = sn.Prepare(ctx, layerName, parentSnapshot, sopts...)
 	}
 
 	if err != nil {
diff --git a/integration/container/create_test.go b/integration/container/create_test.go
index 8cf8f78e06e71..b9f9e3d6317e7 100644
--- a/integration/container/create_test.go
+++ b/integration/container/create_test.go
@@ -21,6 +21,7 @@ import (
 	"github.com/docker/docker/oci"
 	"github.com/docker/docker/pkg/stringid"
 	"github.com/docker/docker/testutil"
+	"github.com/docker/go-units"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"gotest.tools/v3/assert"
 	is "gotest.tools/v3/assert/cmp"
@@ -790,3 +791,52 @@ func TestContainerdContainerImageInfo(t *testing.T) {
 		assert.Equal(t, ctr.Image, "")
 	}
 }
+
+func TestContainerdSnapshotQuota(t *testing.T) {
+	skip.If(t, testEnv.DaemonInfo.OSType != "windows", "Only test windows")
+
+	ctx := setupTest(t)
+
+	apiClient := testEnv.APIClient()
+	defer apiClient.Close()
+
+	info, err := apiClient.Info(ctx)
+	assert.NilError(t, err)
+
+	skip.If(t, info.Containerd == nil, "requires containerd")
+	skip.If(t, info.Driver != "windows", "requires windows daemon")
+	foundSnapshotter := false
+	for _, pair := range info.DriverStatus {
+		if pair[0] == "driver-type" && pair[1] == "io.containerd.snapshotter.v1" {
+			foundSnapshotter = true
+		}
+	}
+	skip.If(t, !foundSnapshotter, "requires snapshotter driver")
+
+	const containerSize = "32G"
+	expectedSize, err := units.RAMInBytes(containerSize)
+	assert.NilError(t, err)
+	expectedSizeStr := strconv.FormatInt(expectedSize, 10)
+
+	id := testContainer.Create(ctx, t, apiClient, func(cfg *testContainer.TestContainerConfig) {
+		cfg.Config.Image = "mcr.microsoft.com/windows/servercore:ltsc2022"
+		cfg.Config.Cmd = []string{"powershell", "-Command", "Start-Sleep -Seconds 20"}
+		cfg.HostConfig.StorageOpt = make(map[string]string)
+		cfg.HostConfig.StorageOpt["size"] = containerSize
+	})
+	defer apiClient.ContainerRemove(ctx, id, container.RemoveOptions{Force: true})
+
+	err = apiClient.ContainerStart(ctx, id, container.StartOptions{})
+	assert.NilError(t, err)
+
+	c8dClient, err := containerd.New(info.Containerd.Address, containerd.WithDefaultNamespace(info.Containerd.Namespaces.Containers))
+	assert.NilError(t, err)
+	defer c8dClient.Close()
+
+	snapshotInfo, err := c8dClient.SnapshotService("windows").Stat(ctx, id)
+	assert.NilError(t, err)
+
+	rootfsLabel, ok := snapshotInfo.Labels["containerd.io/snapshot/windows/rootfs.sizebytes"]
+	assert.Assert(t, ok, "Snapshot does not cotain rootfs quota label")
+	assert.Equal(t, rootfsLabel, expectedSizeStr, "Container snapshot size does not match")
+}