Short answer: accept each seller catalog import once, turn every source image into an independently retryable smart-crop item, and report progress from durable item states rather than from a worker's estimate. For an edtech marketplace producing several lesson-card aspect ratios, storage and cache traffic usually deserve the first design pass: keep one recoverable source, generate only the crop variants that a named surface consumes, and put a retention deadline on everything else.
This is a cost-control problem disguised as an upload endpoint. The bill contains source bytes retained over time, derived bytes retained over time, transformation work, requests, and bytes delivered through the cache. Before choosing a queue or image library, write the model down:
monthly cost = source storage + derived storage + crop work + cache requests + cache misses
The dominant term can't be inferred from image count alone. A course tile repeatedly opened by students has a different profile from a seller's draft image that is imported once and rejected. Measure bytes and request counts by lifecycle state, surface, and aspect ratio; then remove the retained artifact or repeated transformation that actually drives the term.
What should seller catalog import batch progress measure?
Progress should count committed item transitions, not elapsed time and not messages read from a queue. Give each submitted image a stable item ID, record one row per required rendition, and derive the batch view from those rows. An item is accepted, running, succeeded, or failed; the batch is complete only when every item is in a terminal state. This makes a retry boring, which is exactly what an import path needs.
The exactly-once mindset belongs at the effect boundary. A message can be delivered again, so the worker claims an item conditionally and writes its output key plus terminal state in the same durable operation, or in operations joined by an outbox. The uniqueness rule is (import_id, item_id, ratio). If that identity already has a successful result, the worker returns it instead of writing another object. Don't increment a free-standing completed counter and hope it agrees with the item ledger after a crash.
Use an idempotency key for submission as well. The key maps to a digest of the normalized request and the resulting import ID. Repeating the same key and same digest returns the original acceptance response; repeating the key with different content returns 409 Conflict. That response is an audit signal, not an invitation to silently replace a seller's catalog.
Here is the state contract used by the example below:
| Field | Meaning | Audit rule |
|---|---|---|
import_id |
Stable batch identity | Never reused for different request content |
item_id |
Seller-controlled image identity | Unique within the import |
ratio |
Required display shape | Comes from an allowlist owned by product surfaces |
state |
Committed processing state | Monotonic except an explicit retry from failed
|
output_key |
Address of a finished rendition | Written only with succeeded
|
updated_at |
Last committed transition | Server-generated and retained in the audit log |
Keep the denominator fixed after acceptance.
If validation discovers an unsupported media type, record a failed item rather than deleting it from the batch total; otherwise the displayed percentage can move backward or make a partial import look complete. The MDN media format guide is useful when defining the accepted-format policy, but browser support and decoder policy are separate decisions. Decode untrusted input in an isolated worker, enforce explicit byte and pixel limits, and preserve the rejection reason without preserving the rejected payload indefinitely.
Submit once, then read a durable projection
The following program is intentionally small, but it is runnable: save it as main.go, run go run main.go, submit JSON to POST /imports, and poll the returned path with GET. It uses memory only, so process durability is outside its boundary; the interfaces and transition rules are the part to carry into a transactional store and queue.
package main
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"log"
"net/http"
"strings"
"sync"
"time"
)
type Item struct {
ID string `json:"item_id"`
SourceKey string `json:"source_key"`
Ratios []string `json:"ratios"`
}
type Submission struct {
SellerID string `json:"seller_id"`
Items []Item `json:"items"`
}
type Work struct {
ItemID string `json:"item_id"`
Ratio string `json:"ratio"`
State string `json:"state"`
OutputKey string `json:"output_key,omitempty"`
ErrorCode string `json:"error_code,omitempty"`
}
type Import struct {
ID string `json:"import_id"`
State string `json:"state"`
Total int `json:"total"`
Succeeded int `json:"succeeded"`
Failed int `json:"failed"`
UpdatedAt time.Time `json:"updated_at"`
Work []Work `json:"work"`
}
type idempotencyRecord struct {
Digest string
ImportID string
}
type store struct {
mu sync.RWMutex
imports map[string]*Import
keys map[string]idempotencyRecord
}
func newStore() *store {
return &store{imports: map[string]*Import{}, keys: map[string]idempotencyRecord{}}
}
var allowedRatios = map[string]bool{"1:1": true, "4:3": true, "16:9": true}
func digest(body []byte) string {
sum := sha256.Sum256(body)
return hex.EncodeToString(sum[:])
}
func shortID(value string) string {
sum := sha256.Sum256([]byte(value))
return hex.EncodeToString(sum[:8])
}
func validate(in Submission) error {
if strings.TrimSpace(in.SellerID) == "" || len(in.Items) == 0 {
return errors.New("seller_id and at least one item are required")
}
seen := map[string]bool{}
for _, item := range in.Items {
if item.ID == "" || item.SourceKey == "" || len(item.Ratios) == 0 {
return errors.New("each item needs item_id, source_key, and ratios")
}
if seen[item.ID] {
return fmt.Errorf("duplicate item_id %q", item.ID)
}
seen[item.ID] = true
for _, ratio := range item.Ratios {
if !allowedRatios[ratio] {
return fmt.Errorf("unsupported ratio %q", ratio)
}
}
}
return nil
}
func (s *store) submit(key string, body []byte, in Submission) (*Import, bool, error) {
s.mu.Lock()
defer s.mu.Unlock()
d := digest(body)
if prior, ok := s.keys[key]; ok {
if prior.Digest != d {
return nil, false, errors.New("idempotency key reused with different content")
}
return s.imports[prior.ImportID], true, nil
}
job := &Import{ID: shortID(key), State: "accepted", UpdatedAt: time.Now().UTC()}
for _, item := range in.Items {
for _, ratio := range item.Ratios {
job.Work = append(job.Work, Work{ItemID: item.ID, Ratio: ratio, State: "accepted"})
}
}
job.Total = len(job.Work)
s.imports[job.ID] = job
s.keys[key] = idempotencyRecord{Digest: d, ImportID: job.ID}
return job, false, nil
}
func (s *store) complete(importID string) {
s.mu.Lock()
defer s.mu.Unlock()
job := s.imports[importID]
if job == nil {
return
}
job.State = "running"
for i := range job.Work {
if job.Work[i].State != "accepted" {
continue
}
job.Work[i].State = "succeeded"
job.Work[i].OutputKey = fmt.Sprintf("renditions/%s/%s/%s", job.ID, job.Work[i].ItemID, job.Work[i].Ratio)
job.Succeeded++
job.UpdatedAt = time.Now().UTC()
}
job.State = "succeeded"
}
func main() {
s := newStore()
mux := http.NewServeMux()
mux.HandleFunc("POST /imports", func(w http.ResponseWriter, r *http.Request) {
key := r.Header.Get("Idempotency-Key")
if key == "" {
http.Error(w, "Idempotency-Key is required", http.StatusBadRequest)
return
}
var in Submission
dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20))
if err := dec.Decode(&in); err != nil {
http.Error(w, "invalid JSON", http.StatusBadRequest)
return
}
if err := validate(in); err != nil {
http.Error(w, err.Error(), http.StatusUnprocessableEntity)
return
}
body, _ := json.Marshal(in)
job, replay, err := s.submit(key, body, in)
if err != nil {
http.Error(w, err.Error(), http.StatusConflict)
return
}
if !replay {
go s.complete(job.ID)
}
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Location", "/imports/"+job.ID)
w.WriteHeader(http.StatusAccepted)
json.NewEncoder(w).Encode(map[string]string{"import_id": job.ID, "status_url": "/imports/" + job.ID})
})
mux.HandleFunc("GET /imports/{id}", func(w http.ResponseWriter, r *http.Request) {
s.mu.RLock()
job := s.imports[r.PathValue("id")]
if job == nil {
s.mu.RUnlock()
http.NotFound(w, r)
return
}
copy := *job
copy.Work = append([]Work(nil), job.Work...)
s.mu.RUnlock()
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(copy)
})
log.Fatal(http.ListenAndServe(":8080", mux))
}
The complete method stands in for a crop worker; production code should call a decoder and cropper through an interface, then commit each rendition separately. It must not mark the whole import successful after the first output. The in-memory lock demonstrates atomic claims within one process, not across replicas, and the request-body limit is merely an example for metadata because the source images are referenced by key rather than uploaded through this JSON endpoint.
There is one subtle trap in this sample: canonical JSON is created after decoding, so harmless member ordering differences replay cleanly, but array order remains significant. Decide whether item order is part of the request identity. I'm not sure there is one right answer for every catalog; preserving order helps audit the exact seller submission, while sorting items can make client retries more forgiving. A contract test with reordered items should force that decision before deployment.
Make smart-crop cost a schema decision
Suppose the product team asks for 1:1, 4:3, and 16:9 renditions for every source. Do not translate that request directly into three permanent objects. First map each ratio to a real surface and record its expected access pattern. If 4:3 belongs only to an unpublished authoring preview, an expiring rendition or an on-demand crop may be cheaper than permanent storage; if 16:9 appears on every course landing page, precomputing it can avoid repeated crop work and improve cache reuse. These are hypotheses until telemetry supplies byte-months, transformations, requests, misses, and evictions for each class.
Use an illustrative worksheet rather than a claimed benchmark. For N accepted sources, average source size S, rendition sizes R1..Rk, retention fractions f1..fk, and cache-miss counts m1..mk, retained derived bytes are N x sum(Ri x fi). If one ratio has a low fi because few sellers publish it, generating it only after publication removes (1 - fi) x N x Ri retained bytes, while potentially adding crop latency and work on its first request. Plug in observed values from your own workload. Your mileage may vary — especially when a small set of popular lessons dominates delivery.
The cache key must include the immutable source version, crop policy version, ratio, focal-point inputs, encoder settings, and output format. Without those fields, a policy rollout can serve an old crop under a new catalog record; with mutable URLs and broad invalidation, it can also destroy cache reuse. Prefer content-addressed or versioned output keys, switch the catalog pointer after the new rendition succeeds, and let old keys expire under policy. This is the media equivalent of posting a ledger entry instead of editing history.
Short-lived variants are the first thing to stop keeping. Retain the accepted source while recropping remains a product requirement, retain published renditions while catalog records reference them, and expire preview renditions, rejected uploads, abandoned batch inputs, and superseded outputs according to documented policy. The catch is recovery: deleting an original removes the ability to reproduce a crop after a policy change, while deleting derived objects means the first later read must regenerate them. Storage saved now is purchased with recovery latency or irreversible loss later.
Observe transitions, reconciliation, and retention
Expose counts and item-level failures in the status response, but build operations around the transition log. Each transition should carry the import ID, item identity, prior and next states, attempt number, policy version, timestamps, and a stable error code. Keep seller-provided text out of metric labels; cardinality and data-handling rules both argue for structured logs plus bounded labels.
Reconciliation is mandatory.
Periodically compare accepted work with terminal work, successful rows with existing output objects, and catalog pointers with their referenced rendition keys. A difference does not authorize automatic deletion. Emit a reconciliation finding, classify it, and apply a reviewed repair operation with its own idempotency key and audit record. Fast polling may make a dashboard look alive, but it can't prove that storage and database state agree.
Compliance requirements vary by jurisdiction, contract, and the type of material sellers upload, so the retention schedule needs review by the people responsible for those obligations. Engineering should make the policy executable: store a deletion deadline and policy reason, record deletion completion without retaining the deleted payload, and test legal-hold or investigation controls if those apply. Don't promise that a cache purge alone deletes every retained copy unless the cache contract and your verification process support that statement.
Deployment should begin with shadow accounting: calculate desired crops and projected retention without writing new renditions, then compare the model with current object and cache telemetry. Roll out one surface and one ratio, reconcile it, and expand only after terminal counts and stored objects balance. Alerts should cover old nonterminal items, repeated attempts, reconciliation mismatches, deletion backlog, and a sudden cache-miss change by policy version.
This design is not suitable when the caller requires all transformed bytes in the original synchronous response, or when the batch is so small and infrequent that operating a durable queue and transition ledger costs more than it controls. In that case, keep a synchronous transform path with the same idempotency and audit identities. For large seller imports, though, the asynchronous ledger earns its complexity because submission, processing, visibility, and recovery no longer share one failure window.
Test the invariants before optimizing throughput
Test duplicate submission with the same key, duplicate submission with changed content, repeated worker delivery, worker interruption between object creation and state commit, policy-version changes, partial terminal batches, and retention deletion. The strongest assertions are accounting assertions: total = accepted + running + succeeded + failed, every successful rendition has one output identity, and every visible catalog pointer resolves to the expected policy version.
Three words: trust the ledger.
Load tests should preserve the workload's skew rather than send uniformly random images. Measure queue age, transition commit latency, crop duration by input class, retained bytes by lifecycle, and cache misses by rendition policy. Throughput matters, but a faster worker that creates unused permanent variants can increase the dominant cost.
The stopping rule is concrete: ship a ratio only when a named surface consumes it; keep a source only while recovery or recropping policy requires it; treat progress as a projection of auditable item states; and reconcile object storage against that record. When something goes wrong after aggressive deletion, accept the declared trade-off: regenerate a derived rendition from the retained source, or report that the crop cannot be reproduced if the source's retention window has closed. Never invent progress to hide that boundary.
Top comments (0)