Firehose Recordの任意集約と容量・再送・drain制御を追加する

README.md

@@ -37,8 +37,40 @@ func main() {
 ```
 
 ## Description
-The firequeue utilizes an in-memory queue to ensure input to Amazon Kinesis Data Firehose.
-When the looping process is cancelled by the context, the firequeue wait for the queue to be empty and then exit the loop.
+firequeue owns an in-memory queue. `Enqueue` copies payloads; acceptance does not
+guarantee persistence. Stop producers before cancelling `Loop`'s context. The queue
+then rejects new input and drains with a fresh context, for at most 10 seconds by
+default (`DrainTimeout` changes this). Always inspect `Loop`'s returned error: a
+timeout reports the remaining event and payload-byte counts. Process loss can lose
+pending data; ambiguous request outcomes can duplicate records.
+
+By default each input remains a separate record. For newline-delimited JSON, callers
+can opt into concatenation without re-encoding or adding delimiters:
+
+```go
+fq := firequeue.New(fh, "device-messages",
+    firequeue.Aggregate(64*1024, 500*time.Millisecond),
+)
+```
+
+The first event sets the aggregation deadline; later arrivals do not extend it.
+Payloads larger than the target remain standalone. Delivery backlog and retries
+can add latency after sealing. Each payload must contain 1–1,024,000 bytes. Each
+request contains at most 500 records and 4,000,000 payload bytes, before Base64.
+
+`MaxQueueLength` (default 100,000 events) and `MaxQueueBytes` (default 64 MiB of
+payload, excluding allocation overhead) include aggregation, ready, retry, and
+in-flight data. Capacity rejection is explicit. HTTP-success partial failures retry
+only failed records with the same bytes. Invalid responses and ambiguous transport
+failures retry all affected records. Retryable requests use exponential backoff.
+
+`Stats` separates input events from attempted/acknowledged Firehose records.
+`AcceptedEvents = SuccessfulEvents + FailedEvents + PendingEvents`.
+Terminal API failures invoke `ErrorHandler` and increment `FailedEvents`; `Loop`
+can still return nil once the queue is empty. Estimated billable bytes round each
+attempt up to 5 KiB and do not represent an AWS invoice or SDK-internal retries.
+`AggregationWaitObserver` measures creation-to-seal time once per record, when the
+first send attempt starts; callbacks must be concurrency-safe and return promptly.
 
 ## Installation
 

delivery.go

@@ -0,0 +1,130 @@
+package firequeue
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/aws/request"
+	"github.com/aws/aws-sdk-go/service/firehose"
+)
+
+const requestTimeout = 30 * time.Second
+
+func (q *Queue) send(ctx context.Context, batch []*queuedRecord) bool {
+	records := make([]*firehose.Record, len(batch))
+	var billableBytes int64
+	for i, record := range batch {
+		records[i] = record.record
+		billableBytes += int64((len(record.record.Data) + billingUnitBytes - 1) / billingUnitBytes * billingUnitBytes)
+	}
+	q.mu.Lock()
+	q.stats.BatchLength = int64(len(batch))
+	q.stats.SentRecords += int64(len(batch))
+	q.stats.EstimatedBillableBytes += billableBytes
+	q.mu.Unlock()
+	requestCtx, cancel := context.WithTimeout(ctx, requestTimeout)
+	defer cancel()
+	resp, err := q.firehose.PutRecordBatchWithContext(requestCtx, &firehose.PutRecordBatchInput{
+		DeliveryStreamName: q.DeliveryStreamName,
+		Records:            records,
+	})
+	if err != nil {
+		if ctx.Err() != nil || requestCtx.Err() != nil || isRetryable(err) {
+			q.retry(batch)
+			if ctx.Err() == nil {
+				q.handleError(err, records)
+			}
+			return true
+		}
+		q.finish(batch, false)
+		q.handleError(err, records)
+		return false
+	}
+	if err := validateResponse(resp, len(batch)); err != nil {
+		q.retry(batch)
+		q.handleError(err, records)
+		return true
+	}
+	var succeeded, failed []*queuedRecord
+	for i, record := range batch {
+		if resp.RequestResponses[i].ErrorCode != nil && *resp.RequestResponses[i].ErrorCode != "" {
+			failed = append(failed, record)
+		} else {
+			succeeded = append(succeeded, record)
+		}
+	}
+	q.finish(succeeded, true)
+	q.retry(failed)
+	return len(failed) > 0
+}
+
+func validateResponse(resp *firehose.PutRecordBatchOutput, count int) error {
+	if resp == nil || resp.FailedPutCount == nil || len(resp.RequestResponses) != count {
+		return fmt.Errorf("invalid PutRecordBatch response for %d records", count)
+	}
+	var failed int64
+	for _, entry := range resp.RequestResponses {
+		if entry == nil {
+			return fmt.Errorf("nil PutRecordBatch response entry")
+		}
+		if entry.ErrorCode != nil && *entry.ErrorCode != "" {
+			failed++
+		} else if entry.RecordId == nil || *entry.RecordId == "" {
+			return fmt.Errorf("missing PutRecordBatch acknowledgment ID")
+		}
+	}
+	if failed != *resp.FailedPutCount {
+		return fmt.Errorf("inconsistent PutRecordBatch response: %d failures, declared %d", failed, *resp.FailedPutCount)
+	}
+	return nil
+}
+
+func (q *Queue) finish(batch []*queuedRecord, succeeded bool) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	for _, record := range batch {
+		delete(q.inFlight, record)
+		q.stats.PendingEvents -= record.events
+		q.stats.PendingBytes -= int64(len(record.record.Data))
+		if succeeded {
+			q.stats.Success++
+			q.stats.SuccessfulEvents += record.events
+		} else {
+			q.stats.UnretryableError++
+			q.stats.FailedEvents += record.events
+		}
+	}
+}
+
+func (q *Queue) retry(batch []*queuedRecord) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	for _, record := range batch {
+		delete(q.inFlight, record)
+		q.stats.RetryCount++
+		q.queue = append(q.queue, record)
+	}
+}
+
+func (q *Queue) handleError(err error, records []*firehose.Record) {
+	if q.errorHandler != nil {
+		q.errorHandler(err, records)
+		return
+	}
+	log.Println(err)
+}
+
+func isRetryable(err error) bool {
+	if failure, ok := err.(awserr.RequestFailure); ok {
+		switch failure.StatusCode() {
+		case 429, 500, 502, 503, 504:
+			return true
+		}
+	}
+	return request.IsErrorRetryable(err) || request.IsErrorThrottle(err) ||
+		strings.Contains(err.Error(), "read: connection reset by peer")
+}