Artifact [bfd1ff0955]

Artifact bfd1ff0955475ac1422c496e56f4ebbcf2d4fc600119513115944132a6637eda:

• File internal/events/counter.go — part of check-in [dc058fa551] at 2026-08-24 19:16:15 on branch clone-counts — Count clones and fetches

  A clone reads and writes nothing, so it is the one kind of forge activity that leaves no trace in the DAG at all.

  The sync edge classifies the protocol's cards: clone SEQNO 1 opens a new clone, later rounds do not, so a big store cloned in forty rounds still counts once. Two sinks: the vocabulary event for anything reacting, and a JetStream counter for how many, which converges across nodes instead of losing increments to a read-modify-write. (user: dmestas, size: 4769) [annotate] [blame] [check-ins using] [more...]

package events

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// CounterStream holds monotone tallies — clones, fetches, anything the
// forge counts rather than records.
//
// It is a JetStream counter stream, which is a PN-counter CRDT: the
// server keeps the running total and a per-source slot, so two nodes
// incrementing concurrently converge instead of clobbering, and a retry
// after a timeout cannot double-count. That last property is the reason
// this is not a SELECT/UPDATE in the OLTP — a sync round that is retried
// by a flaky client must not add two.
//
// What it deliberately does NOT do is allocate: a counter converges on a
// TOTAL, never on "which number is mine". Anything needing a unique
// value wants KV's revision-checked update instead.
const (
	CounterStream  = "QUARRYCOUNT"
	counterSubject = "quarry.count.>"
	incrHeader     = "Nats-Incr"
)

// counterConfig. Counter streams are Limits-retention only, and the
// server refuses to turn AllowMsgCounter off once on, so this shape is
// permanent for the stream's life. MaxMsgsPerSubject 1 keeps only the
// running total per counter: the history is the event stream's job
// (RFC 0005), and a tally that kept every increment would grow without
// bound to say a number it already knows.
//
// Declared against the newer jetstream API because AllowMsgCounter
// exists only there; everything else in this package speaks the legacy
// JetStreamContext, and both ride the same connection.
var counterConfig = jetstream.StreamConfig{
	Name:              CounterStream,
	Subjects:          []string{counterSubject},
	Retention:         jetstream.LimitsPolicy,
	AllowMsgCounter:   true,
	AllowDirect:       true,
	MaxMsgsPerSubject: 1,
}

// CounterSubject is where a repo's tally of one kind lives.
func CounterSubject(repo, name string) string {
	return "quarry.count." + repo + "." + name
}

// Count adds delta to a tally.
//
// Best-effort like Raise: a lost increment costs one off a statistic,
// where failing the operation that triggered it would cost a clone. The
// new total is deliberately not returned — reading it back is Total's
// job, and a caller that only increments should not be made to care.
func (s *Server) Count(repo, name string, delta int64) error {
	if s == nil {
		return nil
	}
	if repo == "" || name == "" {
		return fmt.Errorf("events: a counter needs a repo and a name")
	}
	m := nats.NewMsg(CounterSubject(repo, name))
	// The sign is required by the server's own regex: +1, never 1.
	m.Header.Set(incrHeader, fmt.Sprintf("%+d", delta))
	if _, err := s.js.PublishMsg(m); err != nil {
		return fmt.Errorf("events: count %s/%s: %w", repo, name, err)
	}
	return nil
}

// Total reads one tally. A counter that has never been incremented is
// zero rather than an error: "nobody has cloned this yet" is an answer,
// not a failure.
func (s *Server) Total(repo, name string) (int64, error) {
	if s == nil {
		return 0, nil
	}
	msg, err := s.js.GetLastMsg(CounterStream, CounterSubject(repo, name))
	if err != nil {
		if errors.Is(err, nats.ErrMsgNotFound) || errors.Is(err, nats.ErrStreamNotFound) {
			return 0, nil
		}
		return 0, fmt.Errorf("events: total %s/%s: %w", repo, name, err)
	}
	return parseCounter(msg.Data)
}

// parseCounter reads the server's body form, {"val":"<total>"}. The
// total is a STRING in the JSON because the server uses big.Int and a
// JSON number would lose precision past 2^53 — so it is parsed as one
// rather than as a number that happens to look right for a while.
func parseCounter(body []byte) (int64, error) {
	var v struct {
		Val string `json:"val"`
	}
	if err := json.Unmarshal(body, &v); err != nil {
		return 0, fmt.Errorf("events: counter body %q: %w", body, err)
	}
	if v.Val == "" {
		return 0, nil
	}
	n, err := strconv.ParseInt(v.Val, 10, 64)
	if err != nil {
		return 0, fmt.Errorf("events: counter value %q: %w", v.Val, err)
	}
	return n, nil
}

// addCounterStream creates the tally stream through the newer API. An
// existing stream is left exactly as it is: AllowMsgCounter cannot be
// turned off once on, so a config update here could only ever fail or
// be a no-op, and failing would take down a server whose counters were
// working fine.
func addCounterStream(nc *nats.Conn) error {
	js, err := jetstream.New(nc)
	if err != nil {
		return fmt.Errorf("events: jetstream: %w", err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if _, err := js.CreateStream(ctx, counterConfig); err != nil {
		if errors.Is(err, jetstream.ErrStreamNameAlreadyInUse) {
			return nil
		}
		return fmt.Errorf("events: counter stream: %w", err)
	}
	return nil
}