func retryAfterDecompression(conn pgxconn.PgxConn, req copyRequest, minTimeInt int64) error {
	var (
		table   = req.table
		minTime = model.Time(minTimeInt).Time()
	)
	ingestSpeedup := 2
	delayBy := (time.Since(minTime) / time.Duration(ingestSpeedup)) + 60*time.Minute
	maxDelayBy := time.Hour * 24
	if delayBy > maxDelayBy {
		delayBy = maxDelayBy
	}
	log.Warn("msg", fmt.Sprintf("Table %s was compressed, decompressing", table), "table", table, "min-time", minTime, "age", time.Since(minTime), "delay-job-by", delayBy)
	_, rescheduleErr := conn.Exec(context.Background(), "SELECT "+schema.Catalog+".delay_compression_job($1, $2)",
		table, time.Now().Add(delayBy))
	if rescheduleErr != nil {
		log.Error("msg", rescheduleErr, "context", "Rescheduling compression")
		return rescheduleErr
	}
	_, decompressErr := conn.Exec(context.Background(), "CALL "+schema.Catalog+".decompress_chunks_after($1, $2);", table, minTime)
	if decompressErr != nil {
		log.Error("msg", decompressErr, "context", "Decompressing chunks")
		return decompressErr
	}
	metrics.DecompressCalls.Inc()
	metrics.DecompressEarliest.WithLabelValues(table).Set(float64(minTime.UnixNano()) / 1e9)
	err, _ := insertSeries(conn, req)  
	return err
}
In the event we filling in old data and the chunk we want to INSERT into has
already been compressed, we decompress the chunk and try again. When we do
this we delay the recompression to give us time to insert additional data.
package ingestor
import (
	"context"
	"fmt"
	"math"
	"sort"
	"strings"
	"sync"
	"time"
	"github.com/jackc/pgconn"
	"github.com/prometheus/common/model"
	"github.com/timescale/promscale/pkg/log"
	"github.com/timescale/promscale/pkg/pgmodel/common/schema"
	"github.com/timescale/promscale/pkg/pgmodel/metrics"
	pgmodel "github.com/timescale/promscale/pkg/pgmodel/model"
	"github.com/timescale/promscale/pkg/pgxconn"
)
const maxInsertStmtPerTxn = 100
type copyRequest struct {
	data  *pendingBuffer
	table string
}
var (
	getBatchMutex       = &sync.Mutex{}
	handleDecompression = retryAfterDecompression
)
type copyBatch []copyRequest
func (reqs copyBatch) VisitSeries(callBack func(s *pgmodel.Series) error) error {
	for _, req := range reqs {
		insertables := req.data.batch.Data()
		for i := range insertables {
			if err := callBack(insertables[i].Series()); err != nil {
				return err
			}
		}
	}
	return nil
}
func (reqs copyBatch) VisitExemplar(callBack func(s *pgmodel.PromExemplars) error) error {
	for _, req := range reqs {
		insertables := req.data.batch.Data()
		for i := range insertables {
			exemplar, ok := insertables[i].(*pgmodel.PromExemplars)
			if ok {
				if err := callBack(exemplar); err != nil {
					return err
				}
			}
		}
	}
	return nil
}
func runCopier(conn pgxconn.PgxConn, in chan readRequest, sw *seriesWriter, elf *ExemplarLabelFormatter) {
	requestBatch := make([]readRequest, 0, maxInsertStmtPerTxn)
	insertBatch := make([]copyRequest, 0, cap(requestBatch))
	for {
		var ok bool
		requestBatch, ok = copierGetBatch(requestBatch, in)
		if !ok {
			return
		}
		for i := range requestBatch {
			copyRequest, ok := <-requestBatch[i].copySender
			if !ok {
				continue
			}
			insertBatch = append(insertBatch, copyRequest)
		}
		sort.Slice(insertBatch, func(i, j int) bool {
			return insertBatch[i].table < insertBatch[j].table
		})
		err := persistBatch(conn, sw, elf, insertBatch)
		if err != nil {
			for i := range insertBatch {
				insertBatch[i].data.reportResults(err)
				insertBatch[i].data.release()
			}
		}
		for i := range requestBatch {
			requestBatch[i] = readRequest{}
		}
		for i := range insertBatch {
			insertBatch[i] = copyRequest{}
		}
		insertBatch = insertBatch[:0]
		requestBatch = requestBatch[:0]
	}
}
func persistBatch(conn pgxconn.PgxConn, sw *seriesWriter, elf *ExemplarLabelFormatter, insertBatch []copyRequest) error {
	batch := copyBatch(insertBatch)
	err := sw.WriteSeries(batch)
	if err != nil {
		return fmt.Errorf("copier: writing series: %w", err)
	}
	err = elf.orderExemplarLabelValues(batch)
	if err != nil {
		return fmt.Errorf("copier: formatting exemplar label values: %w", err)
	}
	doInsertOrFallback(conn, insertBatch...)
	return nil
}
func copierGetBatch(batch []readRequest, in <-chan readRequest) ([]readRequest, bool) {
	getBatchMutex.Lock()
	defer getBatchMutex.Unlock()
	req, ok := <-in
	if !ok {
		return batch, false
	}
	batch = append(batch, req)
	timeout := time.After(20 * time.Millisecond)
hot_gather:
	for len(batch) < cap(batch) {
		select {
		case r2 := <-in:
			batch = append(batch, r2)
		case <-timeout:
			break hot_gather
		}
	}
	return batch, true
}
func doInsertOrFallback(conn pgxconn.PgxConn, reqs ...copyRequest) {
	err, _ := insertSeries(conn, reqs...)
	if err != nil {
		insertBatchErrorFallback(conn, reqs...)
		return
	}
	for i := range reqs {
		reqs[i].data.reportResults(nil)
		reqs[i].data.release()
	}
}
func insertBatchErrorFallback(conn pgxconn.PgxConn, reqs ...copyRequest) {
	for i := range reqs {
		err, minTime := insertSeries(conn, reqs[i])
		if err != nil {
			err = tryRecovery(conn, err, reqs[i], minTime)
		}
		reqs[i].data.reportResults(err)
		reqs[i].data.release()
	}
}
func tryRecovery(conn pgxconn.PgxConn, err error, req copyRequest, minTime int64) error {
	pgErr, ok := err.(*pgconn.PgError)
	if !ok {
		errMsg := err.Error()
		log.Warn("msg", fmt.Sprintf("unexpected error while inserting to %s", req.table), "err", errMsg)
		return err
	}
	if pgErr.Code == "0A000" || strings.Contains(pgErr.Message, "compressed") || strings.Contains(pgErr.Message, "insert/update/delete not permitted") {
		return handleDecompression(conn, req, minTime)
	}
	log.Warn("msg", fmt.Sprintf("unexpected postgres error while inserting to %s", req.table), "err", pgErr.Error())
	return pgErr
}
func skipDecompression(_ pgxconn.PgxConn, _ copyRequest, _ int64) error {
	log.WarnRateLimited("msg", "Rejecting samples falling on compressed chunks as decompression is disabled")
	return nil
}