<commit_before>package worker

import (
	"sync"
	"time"

	"github.com/concourse/baggageclaim"
	"github.com/pivotal-golang/clock"
	"github.com/pivotal-golang/lager"
)

const volumeKeepalive = 30 * time.Second

//go:generate counterfeiter . VolumeFactoryDB

type VolumeFactoryDB interface {
	GetVolumeTTL(volumeHandle string) (time.Duration, error)
	ReapVolume(handle string) error
	SetVolumeTTL(string, time.Duration) error
}

//go:generate counterfeiter . VolumeFactory

type VolumeFactory interface {
	Build(lager.Logger, baggageclaim.Volume) (Volume, error)
}

type volumeFactory struct {
	db    VolumeFactoryDB
	clock clock.Clock
}

func NewVolumeFactory(db VolumeFactoryDB, clock clock.Clock) VolumeFactory {
	return &volumeFactory{
		db:    db,
		clock: clock,
	}
}

func (vf *volumeFactory) Build(logger lager.Logger, bcVol baggageclaim.Volume) (Volume, error) {
	bcVol.Release(nil)
	return newVolume(logger, bcVol, vf.clock, vf.db)
}

//go:generate counterfeiter . Volume

type Volume interface {
	baggageclaim.Volume
}

type volume struct {
	baggageclaim.Volume
	db VolumeFactoryDB

	release      chan *time.Duration
	heartbeating *sync.WaitGroup
	releaseOnce  sync.Once
}

type VolumeMount struct {
	Volume    Volume
	MountPath string
}

func newVolume(logger lager.Logger, bcVol baggageclaim.Volume, clock clock.Clock, db VolumeFactoryDB) (Volume, error) {
	vol := &volume{
		Volume: bcVol,
		db:     db,

		heartbeating: new(sync.WaitGroup),
		release:      make(chan *time.Duration, 1),
	}

	ttl, err := vol.db.GetVolumeTTL(vol.Handle())
	if err != nil {
		logger.Info("failed-to-lookup-ttl", lager.Data{"error": err.Error()})

		ttl, _, err = bcVol.Expiration()
		if err != nil {
			logger.Error("failed-to-lookup-expiration-of-volume", err)
			return nil, err
		}
	}

	vol.heartbeat(logger.Session("initial-heartbeat"), ttl)

	vol.heartbeating.Add(1)
	go vol.heartbeatContinuously(
		logger.Session("continous-heartbeat"),
		clock.NewTicker(volumeKeepalive),
		ttl,
	)

	return vol, nil
}

func (v *volume) Release(finalTTL *time.Duration) {
	v.releaseOnce.Do(func() {
		v.release <- finalTTL
		v.heartbeating.Wait()
	})

	return
}

func (v *volume) heartbeatContinuously(logger lager.Logger, pacemaker clock.Ticker, initialTTL time.Duration) {
	defer v.heartbeating.Done()
	defer pacemaker.Stop()
<commit_msg>fix typo<commit_after>package worker

import (
	"sync"
	"time"

	"github.com/concourse/baggageclaim"
	"github.com/pivotal-golang/clock"
	"github.com/pivotal-golang/lager"
)

const volumeKeepalive = 30 * time.Second

//go:generate counterfeiter . VolumeFactoryDB

type VolumeFactoryDB interface {
	GetVolumeTTL(volumeHandle string) (time.Duration, error)
	ReapVolume(handle string) error
	SetVolumeTTL(string, time.Duration) error
}

//go:generate counterfeiter . VolumeFactory

type VolumeFactory interface {
	Build(lager.Logger, baggageclaim.Volume) (Volume, error)
}

type volumeFactory struct {
	db    VolumeFactoryDB
	clock clock.Clock
}

func NewVolumeFactory(db VolumeFactoryDB, clock clock.Clock) VolumeFactory {
	return &volumeFactory{
		db:    db,
		clock: clock,
	}
}

func (vf *volumeFactory) Build(logger lager.Logger, bcVol baggageclaim.Volume) (Volume, error) {
	bcVol.Release(nil)
	return newVolume(logger, bcVol, vf.clock, vf.db)
}

//go:generate counterfeiter . Volume

type Volume interface {
	baggageclaim.Volume
}

type volume struct {
	baggageclaim.Volume
	db VolumeFactoryDB

	release      chan *time.Duration
	heartbeating *sync.WaitGroup
	releaseOnce  sync.Once
}

type VolumeMount struct {
	Volume    Volume
	MountPath string
}

func newVolume(logger lager.Logger, bcVol baggageclaim.Volume, clock clock.Clock, db VolumeFactoryDB) (Volume, error) {
	vol := &volume{
		Volume: bcVol,
		db:     db,

		heartbeating: new(sync.WaitGroup),
		release:      make(chan *time.Duration, 1),
	}

	ttl, err := vol.db.GetVolumeTTL(vol.Handle())
	if err != nil {
		logger.Info("failed-to-lookup-ttl", lager.Data{"error": err.Error()})

		ttl, _, err = bcVol.Expiration()
		if err != nil {
			logger.Error("failed-to-lookup-expiration-of-volume", err)
			return nil, err
		}
	}

	vol.heartbeat(logger.Session("initial-heartbeat"), ttl)

	vol.heartbeating.Add(1)
	go vol.heartbeatContinuously(
		logger.Session("continuous-heartbeat"),
		clock.NewTicker(volumeKeepalive),
		ttl,
	)

	return vol, nil
}

func (v *volume) Release(finalTTL *time.Duration) {
	v.releaseOnce.Do(func() {
		v.release <- finalTTL
		v.heartbeating.Wait()
	})

	return
}

func (v *volume) heartbeatContinuously(logger lager.Logger, pacemaker clock.Ticker, initialTTL time.Duration) {
	defer v.heartbeating.Done()
	defer pacemaker.Stop()

{'fraction_non_alphanumeric': 0.11929371231696813, 'fraction_numerical': 0.0017226528854435831, 'mean_word_length': 4.477594339622642, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6912870', 'n_tokens_mistral': 1697, 'n_tokens_neox': 1599, 'n_words': 353}