Repository: prometheus/prometheus
File: tsdb/agent/db.go (Go)

Code before:

	if st <= series.lastTs {
		// discard the sample if it's out of order.
		return 0, storage.ErrOutOfOrderST
	}
	// NOTE(bwplotka): This is a bug, as we "commit" pending sample TS as the WAL last TS. It was likely done
	// to satisfy incorrect TestDBStartTimestampSamplesIngestion test. We are leaving it as-is given the planned removal
	// of AppenderV1 as per https://github.com/prometheus/prometheus/issues/17632.
	series.lastTs = st

	// NOTE: always modify pendingSamples and sampleSeries together.
	a.pendingSamples = append(a.pendingSamples, record.RefSample{
		Ref: series.ref,
		T:   st,
		V:   0,
	})
	a.sampleSeries = append(a.sampleSeries, series)

	a.metrics.totalAppendedSamples.WithLabelValues(sampleMetricTypeFloat).Inc()

	return storage.SeriesRef(series.ref), nil
}

// Commit submits the collected samples and purges the batch.
func (a *appenderBase) Commit() error {
	if err := a.log(); err != nil {
		return err
	}

	a.clearData()
	a.appenderPool.Put(a)

	if a.writeNotified != nil {
		a.writeNotified.Notify()
	}
	return nil
}

// log logs all pending data to the WAL.
func (a *appenderBase) log() error {
	a.mtx.RLock()
	defer a.mtx.RUnlock()

	var encoder record.Encoder
	buf := a.bufPool.Get().([]byte)
	defer func() {
		a.bufPool.Put(buf) //nolint:staticcheck
	}()

	if len(a.pendingSeries) > 0 {
		buf = encoder.Series(a.pendingSeries, buf)

Review comment (bug):
The original code (before appender v2) [leaked](https://github.com/prometheus/prometheus/blob/a66c696530e2305566c74871d0509b9fa9c502ac/tsdb/agent/db.go#L1130) the appender on error.

It also explicitly [tested](https://github.com/prometheus/prometheus/blob/a66c696530e2305566c74871d0509b9fa9c502ac/tsdb/agent/db_test.go#L334) Commit after Rollback which is forbidden in the [interface documentation](https://github.com/prometheus/prometheus/blob/a66c696530e2305566c74871d0509b9fa9c502ac/storage/interface.go#L281). Also it will call the commit on an object already returned to the pool.

Reading the [scrape](https://github.com/prometheus/prometheus/blob/a66c696530e2305566c74871d0509b9fa9c502ac/scrape/scrape.go#L1429) code I think there's some cases when we call Rollback twice possibly. E.g. when config is wrong and we stop the scrape loop with forced error, we may error on stale markers, do a Rollback()  and then do another Rollback() in the deferred function.

So in general this automatic pool handling seems a bit dangerous, but we're not making it too much worse I guess and fix a bug.

Code after:

	if st <= series.lastTs {
		// discard the sample if it's out of order.
		return 0, storage.ErrOutOfOrderST
	}
	// NOTE(bwplotka): This is a bug, as we "commit" pending sample TS as the WAL last TS. It was likely done
	// to satisfy incorrect TestDBStartTimestampSamplesIngestion test. We are leaving it as-is given the planned removal
	// of AppenderV1 as per https://github.com/prometheus/prometheus/issues/17632.
	series.lastTs = st

	// NOTE: always modify pendingSamples and sampleSeries together.
	a.pendingSamples = append(a.pendingSamples, record.RefSample{
		Ref: series.ref,
		T:   st,
		V:   0,
	})
	a.sampleSeries = append(a.sampleSeries, series)

	a.metrics.totalAppendedSamples.WithLabelValues(sampleMetricTypeFloat).Inc()

	return storage.SeriesRef(series.ref), nil
}

// Commit submits the collected samples and purges the batch.
func (a *appender) Commit() error {
	defer a.appenderPool.Put(a)
	return a.commit()
}

func (a *appender) Rollback() error {
	defer a.appenderPool.Put(a)
	return a.rollback()
}

func (a *appenderBase) commit() error {
	if err := a.log(); err != nil {
		return err
	}

	a.clearData()

	if a.writeNotified != nil {
		a.writeNotified.Notify()
	}
	return nil
}

// log logs all pending data to the WAL.
func (a *appenderBase) log() error {
	a.mtx.RLock()
	defer a.mtx.RUnlock()