Update go-txfile 0.0.6 (elastic#10289)

Closes: elastic#9874

CHANGELOG.next.asciidoc

@@ -99,6 +99,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Gracefully handle TLS options when enrolling a Beat. {issue}9129[9129]
 - The backing off now implements jitter to better distribute the load. {issue}10172[10172]
 - Fix TLS certificate DoS vulnerability. {pull}10302[10302]
+- Fix panic and file unlock in spool on atomic operation (arm, x86-32). File lock was not released when panic occurs, leading to the beat deadlocking on startup. {pull}10289[10289]
 
 *Auditbeat*
 

NOTICE.txt

@@ -657,8 +657,8 @@ Elasticsearch, B.V. (https://www.elastic.co/).
 
 --------------------------------------------------------------------
 Dependency: github.com/elastic/go-txfile
-Version: v0.0.3
-Revision: 389b527ad365f6fc6cf5fa7e0ba5a2294ad2f3ed
+Version: v0.0.6
+Revision: fa3d87c14381c01aaa1c7901877fb5e483294964
 License type (autodetected): Apache-2.0
 ./vendor/github.com/elastic/go-txfile/LICENSE:
 --------------------------------------------------------------------

libbeat/publisher/queue/spool/inbroker.go

@@ -127,6 +127,9 @@ func (b *inBroker) Producer(cfg queue.ProducerConfig) queue.Producer {
 // run in the same go-routine as the Flush was executed from.
 // Only the (*inBroker).eventLoop triggers a flush.
 func (b *inBroker) onFlush(n uint) {
+	log := b.ctx.logger
+	log.Debug("inbroker: onFlush ", n)
+
 	if n == 0 {
 		return
 	}
@@ -379,7 +382,7 @@ func (b *inBroker) stateWithTimer() bool {
 		b.handleCancel(&req)
 
 	case <-b.timer.C:
-		log.Debug("inbroker (stateWithTimer): flush timeout")
+		log.Debug("inbroker (stateWithTimer): flush timeout", b.bufferedEvents)
 
 		b.timer.Stop(true)
 
@@ -394,7 +397,7 @@ func (b *inBroker) stateWithTimer() bool {
 
 		if b.bufferedEvents > 0 {
 			// flush did not push all events? Restart timer.
-			log.Debug("  inbroker (stateWithTimer): start flush timer")
+			log.Debug("  inbroker (stateWithTimer): start flush timer", b.bufferedEvents)
 			b.timer.Start()
 			break
 		}

libbeat/publisher/queue/spool/outbroker.go

@@ -83,7 +83,18 @@ var errRetry = errors.New("retry")
 
 func newOutBroker(ctx *spoolCtx, qu *pq.Queue, flushTimeout time.Duration) (*outBroker, error) {
 	reader := qu.Reader()
-	avail, err := reader.Available()
+
+	var (
+		avail uint
+		err   error
+	)
+	func() {
+		if err = reader.Begin(); err != nil {
+			return
+		}
+		defer reader.Done()
+		avail, err = reader.Available()
+	}()
 	if err != nil {
 		return nil, err
 	}