| 1 | package hostops |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "io" |
| 7 | |
| 8 | "github.com/carlosframework/platform/internal/geoip" |
| 9 | "github.com/carlosframework/platform/internal/release" |
| 10 | ) |
| 11 | |
| 12 | var _ release.PutStreamer = (*mutationBoundary)(nil) |
| 13 | |
| 14 | // PutFrom is release.PutStreamer, written out so the boundary is never |
| 15 | // bypassed by a streaming write and a 130 MB GeoIP database is never buffered |
| 16 | // through it. Only a create-only put of an exact GeoIP blob key streams: it is |
| 17 | // classified by key (its bytes were hashed by the publisher while staging, and |
| 18 | // a create-only put cannot replace an object), then handed to the raw store. |
| 19 | // Every other key does what release.PutObjectFrom does for a store with no |
| 20 | // PutFrom: a bounded buffer and the ordinary checked Put. |
| 21 | func (s *mutationBoundary) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) { |
| 22 | if !IsGeoipBlobKey(key) || ifMatch != "" { |
| 23 | if size > release.PutFromFallbackMax { |
| 24 | return "", fmt.Errorf("%s: %w", key, release.ErrTooLargeForPut) |
| 25 | } |
| 26 | b, err := io.ReadAll(io.LimitReader(body, size+1)) |
| 27 | if err != nil { |
| 28 | return "", err |
| 29 | } |
| 30 | if int64(len(b)) != size { |
| 31 | return "", fmt.Errorf("%s: body is %d bytes, want %d", key, len(b), size) |
| 32 | } |
| 33 | return s.Put(ctx, key, b, ifMatch) |
| 34 | } |
| 35 | if size <= 0 || size > geoip.MaxBlobBytes { |
| 36 | err := fmt.Errorf("%w: %s: size %d out of range", ErrUnclassified, key, size) |
| 37 | s.report(MutationTarget{Surface: "bucket", Method: "Put", Key: key}, err) |
| 38 | return "", err |
| 39 | } |
| 40 | if err := s.decide(ctx, MutationTarget{Surface: "bucket", Method: "Put", Key: key}); err != nil { |
| 41 | return "", err |
| 42 | } |
| 43 | MarkAppMutationIssued(ctx) |
| 44 | return release.PutObjectFrom(ctx, s.raw, key, body, size, "") |
| 45 | } |
| 46 | |
| 47 | // PutFrom preserves the streaming seam beneath the admission boundary. Only |
| 48 | // immutable deployment GeoIP blobs bypass record-body parsing; app writes |
| 49 | // still use the conditional record checks and the outer mutation boundary. |
| 50 | func (s conditionalRecordStore) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) { |
| 51 | if IsGeoipBlobKey(key) && ifMatch == "" { |
| 52 | return release.PutObjectFrom(ctx, s.raw, key, body, size, ifMatch) |
| 53 | } |
| 54 | if size < 0 || size > release.PutFromFallbackMax { |
| 55 | return "", release.ErrTooLargeForPut |
| 56 | } |
| 57 | b, err := io.ReadAll(io.LimitReader(body, size+1)) |
| 58 | if err != nil { |
| 59 | return "", err |
| 60 | } |
| 61 | if int64(len(b)) != size { |
| 62 | return "", ErrIdentity |
| 63 | } |
| 64 | return s.Put(ctx, key, b, ifMatch) |
| 65 | } |
| 66 | |
amadan