<commit_before>package util

import (
	"fmt"
	"strings"

	"gigawatt-common/pkg/gentle"

	"github.com/cenkalti/backoff"
	"github.com/samuel/go-zookeeper/zk"
)

// CreateP functions similarly to `mkdir -p`.
func CreateP(conn *zk.Conn, path string, data []byte, flags int32, acl []zk.ACL) (zxIds []string, err error) {
	zxIds = []string{}
	pieces := strings.Split(strings.Trim(path, "/"), "/")
	var (
		zxId  string
		soFar string
	)
	for _, piece := range pieces {
		soFar += "/" + piece
		if zxId, err = conn.Create(soFar, data, flags, acl); err != nil && err != zk.ErrNodeExists {
			return
		}
		zxIds = append(zxIds, zxId)
	}
	err = nil // Clear out any potential error state, since if we made it this far we're OK.
	return
}

// MustCreateP will keep trying to create the path until it succeeds.
func MustCreateP(conn *zk.Conn, path string, data []byte, flags int32, acl []zk.ACL, strategy backoff.BackOff) (zxIds []string) {
	var err error
	operation := func() error {
		if zxIds, err = CreateP(conn, path, []byte{}, 0, acl); err != nil {
			return err
		}
		return nil
	}
	gentle.RetryUntilSuccess(fmt.Sprintf("conn=%p MustCreateP", conn), operation, strategy)
	return
}

func MustCreateProtectedEphemeralSequential(conn *zk.Conn, path string, data []byte, acl []zk.ACL, strategy backoff.BackOff) (zxId string) {
	var err error
	operation := func() error {
		if zxId, err = conn.CreateProtectedEphemeralSequential(path, data, acl); err != nil {
			return err
		}
		return nil
	}
	gentle.RetryUntilSuccess(fmt.Sprintf("conn=%p MustCreateProtectedEphemeralSequential", conn), operation, strategy)
	return
}<commit_msg>Protect against infinite failure due to non-existent ZK path.<commit_after>package util

import (
	"fmt"
	"strings"

	"gigawatt-common/pkg/gentle"

	"github.com/cenkalti/backoff"
	"github.com/samuel/go-zookeeper/zk"
)

// CreateP functions similarly to `mkdir -p`.
func CreateP(conn *zk.Conn, path string, data []byte, flags int32, acl []zk.ACL) (zxIds []string, err error) {
	zxIds = []string{}
	pieces := strings.Split(strings.Trim(path, "/"), "/")
	var (
		zxId  string
		soFar string
	)
	for _, piece := range pieces {
		soFar += "/" + piece
		if zxId, err = conn.Create(soFar, data, flags, acl); err != nil && err != zk.ErrNodeExists {
			return
		}
		zxIds = append(zxIds, zxId)
	}
	err = nil // Clear out any potential error state, since if we made it this far we're OK.
	return
}

// MustCreateP will keep trying to create the path until it succeeds.
func MustCreateP(conn *zk.Conn, path string, data []byte, flags int32, acl []zk.ACL, strategy backoff.BackOff) (zxIds []string) {
	var err error
	operation := func() error {
		if zxIds, err = CreateP(conn, path, []byte{}, 0, acl); err != nil {
			return err
		}
		return nil
	}
	gentle.RetryUntilSuccess(fmt.Sprintf("conn=%p MustCreateP", conn), operation, strategy)
	return
}

func MustCreateProtectedEphemeralSequential(conn *zk.Conn, path string, data []byte, acl []zk.ACL, strategy backoff.BackOff) (zxId string) {
	var err error
	operation := func() error {
		if pieces := strings.Split(path, "/"); len(pieces) > 2 {
			if _, err = CreateP(conn, strings.Join(pieces[0:len(pieces)-1], "/"), []byte{}, 0, acl); err != nil {
				return err
			}
		}
		if zxId, err = conn.CreateProtectedEphemeralSequential(path, data, acl); err != nil {
			return err
		}
		return nil
	}
	gentle.RetryUntilSuccess(fmt.Sprintf("conn=%p MustCreateProtectedEphemeralSequential", conn), operation, strategy)
	return
}
{'fraction_non_alphanumeric': 0.14174813110983323, 'fraction_numerical': 0.0040253018976423235, 'mean_word_length': 4.064046579330422, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3778891', 'n_tokens_mistral': 1319, 'n_tokens_neox': 1238, 'n_words': 356}